raft: introduce raft_rpc class

The patch contains a skeleton implementation for the Scylla-specific
Raft RPC module.

It uses `netw::messaging_service` as underlying mechanism to send
RPC messages.

The instance is supposed to be bound to a single raft group.

Signed-off-by: Pavel Solodovnikov <pa.solodovnikov@scylladb.com>

configure.py

@@ -544,6 +544,14 @@ extra_cxxflags = {
     'sanitize': {}
 }
 
+scylla_raft_core = [
+    'raft/raft.cc',
+    'raft/server.cc',
+    'raft/fsm.cc',
+    'raft/progress.cc',
+    'raft/log.cc',
+]
+
 scylla_core = (['database.cc',
                 'absl-flat_hash_map.cc',
                 'table.cc',
@@ -866,7 +874,9 @@ scylla_core = (['database.cc',
                 'service/raft/schema_raft_state_machine.cc',
                 'service/raft/raft_sys_table_storage.cc',
                 'serializer.cc',
-                ] + [Antlr3Grammar('cql3/Cql.g')] + [Thrift('interface/cassandra.thrift', 'Cassandra')]
+                'service/raft/raft_rpc.cc',
+                ] + [Antlr3Grammar('cql3/Cql.g')] + [Thrift('interface/cassandra.thrift', 'Cassandra')] \
+                  + scylla_raft_core
                )
 
 api = ['api/api.cc',
@@ -987,14 +997,7 @@ scylla_tests_dependencies = scylla_core + idls + scylla_tests_generic_dependenci
     'test/lib/random_schema.cc',
 ]
 
-scylla_raft_dependencies = [
-    'raft/raft.cc',
-    'raft/server.cc',
-    'raft/fsm.cc',
-    'raft/progress.cc',
-    'raft/log.cc',
-    'utils/uuid.cc'
-]
+scylla_raft_dependencies = scylla_raft_core + ['utils/uuid.cc']
 
 deps = {
     'scylla': idls + ['main.cc', 'release.cc', 'utils/build_id.cc'] + scylla_core + api + alternator + redis,

service/raft/raft_rpc.cc

@@ -0,0 +1,77 @@
+/*
+ * Copyright (C) 2020 ScyllaDB
+ */
+
+/*
+ * This file is part of Scylla.
+ *
+ * Scylla is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU Affero General Public License as published by
+ * the Free Software Foundation, either version 3 of the License, or
+ * (at your option) any later version.
+ *
+ * Scylla is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public License
+ * along with Scylla.  If not, see <http://www.gnu.org/licenses/>.
+ */
+#include "service/raft/raft_rpc.hh"
+#include "gms/inet_address_serializer.hh"
+#include "serializer_impl.hh"
+#include "message/msg_addr.hh"
+#include "message/messaging_service.hh"
+#include "db/timeout_clock.hh"
+
+raft_rpc::raft_rpc(netw::messaging_service& ms, uint64_t group_id, raft::server_id srv_id)
+    : _group_id(group_id), _server_id(srv_id), _messaging(ms)
+{}
+
+future<> raft_rpc::send_snapshot(raft::server_id id, const raft::install_snapshot& snap) {
+    return _messaging.send_raft_send_snapshot(
+        netw::msg_addr(get_inet_address(id)), db::no_timeout, _server_id, id, snap);
+}
+
+future<> raft_rpc::send_append_entries(raft::server_id id, const raft::append_request& append_request) {
+    return _messaging.send_raft_append_entries(
+        netw::msg_addr(get_inet_address(id)), db::no_timeout, _server_id, id, append_request);
+}
+
+future<> raft_rpc::send_append_entries_reply(raft::server_id id, const raft::append_reply& reply) {
+    return _messaging.send_raft_append_entries_reply(
+        netw::msg_addr(get_inet_address(id)), db::no_timeout, _server_id, id, reply);
+}
+
+future<> raft_rpc::send_vote_request(raft::server_id id, const raft::vote_request& vote_request) {
+    return _messaging.send_raft_vote_request(
+        netw::msg_addr(get_inet_address(id)), db::no_timeout, _server_id, id, vote_request);
+}
+
+future<> raft_rpc::send_vote_reply(raft::server_id id, const raft::vote_reply& vote_reply) {
+    return _messaging.send_raft_vote_reply(
+        netw::msg_addr(get_inet_address(id)), db::no_timeout, _server_id, id, vote_reply);
+}
+
+void raft_rpc::add_server(raft::server_id id, raft::server_info info) {
+    // Parse gms::inet_address from server_info
+    auto in = ser::as_input_stream(bytes_view(info));
+    _server_addresses.emplace(std::pair(id, ser::deserialize(in, boost::type<gms::inet_address>())));
+}
+
+void raft_rpc::remove_server(raft::server_id id) {
+    _server_addresses.erase(id);
+}
+
+future<> raft_rpc::abort() {
+    return make_ready_future<>();
+}
+
+gms::inet_address raft_rpc::get_inet_address(raft::server_id id) const {
+    auto it = _server_addresses.find(id);
+    if (it == _server_addresses.end()) {
+        throw std::runtime_error(format("Destination raft server not found (group id: {}, server id: {})", _group_id, id));
+    }
+    return it->second;
+}

service/raft/raft_rpc.hh

@@ -0,0 +1,50 @@
+/*
+ * Copyright (C) 2020 ScyllaDB
+ */
+
+/*
+ * This file is part of Scylla.
+ *
+ * Scylla is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU Affero General Public License as published by
+ * the Free Software Foundation, either version 3 of the License, or
+ * (at your option) any later version.
+ *
+ * Scylla is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public License
+ * along with Scylla.  If not, see <http://www.gnu.org/licenses/>.
+ */
+#pragma once
+
+#include "raft/raft.hh"
+#include "gms/inet_address.hh"
+#include "message/messaging_service_fwd.hh"
+
+// Scylla-specific implementation of raft RPC module.
+//
+// Uses `netw::messaging_service` as an underlying implementation for
+// actually sending RPC messages.
+class raft_rpc : public raft::rpc {
+    uint64_t _group_id;
+    raft::server_id _server_id;
+    std::unordered_map<raft::server_id, gms::inet_address> _server_addresses;
+    netw::messaging_service& _messaging;
+
+public:
+    explicit raft_rpc(netw::messaging_service& ms, uint64_t group_id, raft::server_id srv_id);
+
+    future<> send_snapshot(raft::server_id server_id, const raft::install_snapshot& snap) override;
+    future<> send_append_entries(raft::server_id id, const raft::append_request& append_request) override;
+    future<> send_append_entries_reply(raft::server_id id, const raft::append_reply& reply) override;
+    future<> send_vote_request(raft::server_id id, const raft::vote_request& vote_request) override;
+    future<> send_vote_reply(raft::server_id id, const raft::vote_reply& vote_reply) override;
+    void add_server(raft::server_id id, raft::server_info info) override;
+    void remove_server(raft::server_id id) override;
+    future<> abort() override;
+    // Map raft server_id to inet_address to be consumed by `messaging_service`
+    gms::inet_address get_inet_address(raft::server_id id) const;
+};