[PATCH v1 04/56] hits: move create_hint_sync_point function to host ids

[Gleb Natapov]

One of its caller is in the RESTful API which gets ips from the user, so
we convert ips to ids inside the API handler using gossiper before
calling the function. We need to deprecate ip based API and move to host
id based.
---
api/api_init.hh | 2 +-
api/hinted_handoff.hh | 3 ++-
db/hints/manager.hh | 2 +-
service/storage_proxy.hh | 2 +-
api/api.cc | 6 +++---
api/hinted_handoff.cc | 13 +++++++------
db/hints/manager.cc | 13 +++----------
main.cc | 2 +-
repair/row_level.cc | 2 +-
service/storage_proxy.cc | 2 +-
10 files changed, 21 insertions(+), 26 deletions(-)

diff --git a/api/api_init.hh b/api/api_init.hh
index cf539344984..fe9e770b089 100644
--- a/api/api_init.hh
+++ b/api/api_init.hh
@@ -124,7 +124,7 @@ future<> set_server_storage_proxy(http_context& ctx, sharded<service::storage_pr
future<> unset_server_storage_proxy(http_context& ctx);
future<> set_server_stream_manager(http_context& ctx, sharded<streaming::stream_manager>& sm);
future<> unset_server_stream_manager(http_context& ctx);
-future<> set_hinted_handoff(http_context& ctx, sharded<service::storage_proxy>& p);
+future<> set_hinted_handoff(http_context& ctx, sharded<service::storage_proxy>& p, sharded<gms::gossiper>& g);
future<> unset_hinted_handoff(http_context& ctx);
future<> set_server_cache(http_context& ctx);
future<> unset_server_cache(http_context& ctx);
diff --git a/api/hinted_handoff.hh b/api/hinted_handoff.hh
index 7a83daa30d5..13b9f5acb65 100644
--- a/api/hinted_handoff.hh
+++ b/api/hinted_handoff.hh
@@ -10,12 +10,13 @@

#include <seastar/core/sharded.hh>
#include "api/api_init.hh"
+#include "gms/gossiper.hh"

namespace service { class storage_proxy; }

namespace api {

-void set_hinted_handoff(http_context& ctx, httpd::routes& r, sharded<service::storage_proxy>& p);
+void set_hinted_handoff(http_context& ctx, httpd::routes& r, sharded<service::storage_proxy>& p, sharded<gms::gossiper>& g);
void unset_hinted_handoff(http_context& ctx, httpd::routes& r);

}
diff --git a/db/hints/manager.hh b/db/hints/manager.hh
index 69d18199ed5..afe7e96648f 100644
--- a/db/hints/manager.hh
+++ b/db/hints/manager.hh
@@ -278,7 +278,7 @@ class manager {
///
/// \param target_eps The list of endpoints the sync point should correspond to. When empty, the function assumes all endpoints.
/// \return Sync point corresponding to the specified endpoints.
- sync_point::shard_rps calculate_current_sync_point(std::span<const gms::inet_address> target_eps) const;
+ sync_point::shard_rps calculate_current_sync_point(std::span<const locator::host_id> target_eps) const;

/// \brief Waits until hint replay reach replay positions described in `rps`.
future<> wait_for_sync_point(abort_source& as, const sync_point::shard_rps& rps);
diff --git a/service/storage_proxy.hh b/service/storage_proxy.hh
index 6689b68c5d8..43ec188f5bf 100644
--- a/service/storage_proxy.hh
+++ b/service/storage_proxy.hh
@@ -708,7 +708,7 @@ class storage_proxy : public seastar::async_sharded_service<storage_proxy>, publ
future<> change_hints_host_filter(db::hints::host_filter new_filter);
const db::hints::host_filter& get_hints_host_filter() const;

- future<db::hints::sync_point> create_hint_sync_point(std::vector<gms::inet_address> target_hosts) const;
+ future<db::hints::sync_point> create_hint_sync_point(std::vector<locator::host_id> target_hosts) const;
future<> wait_for_hint_sync_point(const db::hints::sync_point spoint, clock_type::time_point deadline);

const stats& get_stats() const {
diff --git a/api/api.cc b/api/api.cc
index 879f030b9b0..b0924cf6f26 100644
--- a/api/api.cc
+++ b/api/api.cc
@@ -273,10 +273,10 @@ future<> unset_server_cache(http_context& ctx) {
return ctx.http_server.set_routes([&ctx] (routes& r) { unset_cache_service(ctx, r); });
}

-future<> set_hinted_handoff(http_context& ctx, sharded<service::storage_proxy>& proxy) {
+future<> set_hinted_handoff(http_context& ctx, sharded<service::storage_proxy>& proxy, sharded<gms::gossiper>& g) {
return register_api(ctx, "hinted_handoff",
- "The hinted handoff API", [&proxy] (http_context& ctx, routes& r) {
- set_hinted_handoff(ctx, r, proxy);
+ "The hinted handoff API", [&proxy, &g] (http_context& ctx, routes& r) {
+ set_hinted_handoff(ctx, r, proxy, g);
});
}

diff --git a/api/hinted_handoff.cc b/api/hinted_handoff.cc
index 73cd41bafff..00fc513bf21 100644
--- a/api/hinted_handoff.cc
+++ b/api/hinted_handoff.cc
@@ -14,6 +14,7 @@

#include "gms/inet_address.hh"
#include "service/storage_proxy.hh"
+#include "gms/gossiper.hh"

namespace api {

@@ -21,18 +22,18 @@ using namespace json;
using namespace seastar::httpd;
namespace hh = httpd::hinted_handoff_json;

-void set_hinted_handoff(http_context& ctx, routes& r, sharded<service::storage_proxy>& proxy) {
- hh::create_hints_sync_point.set(r, [&proxy] (std::unique_ptr<http::request> req) -> future<json::json_return_type> {
- auto parse_hosts_list = [] (sstring arg) {
+void set_hinted_handoff(http_context& ctx, routes& r, sharded<service::storage_proxy>& proxy, sharded<gms::gossiper>& g) {
+ hh::create_hints_sync_point.set(r, [&proxy, &g] (std::unique_ptr<http::request> req) -> future<json::json_return_type> {
+ auto parse_hosts_list = [&g] (sstring arg) {
std::vector<sstring> hosts_str = split(arg, ",");
- std::vector<gms::inet_address> hosts;
+ std::vector<locator::host_id> hosts;
hosts.reserve(hosts_str.size());

for (const auto& host_str : hosts_str) {
try {
gms::inet_address host;
host = gms::inet_address(host_str);
- hosts.push_back(host);
+ hosts.push_back(g.local().get_host_id(host));
} catch (std::exception& e) {
throw httpd::bad_param_exception(format("Failed to parse host address {}: {}", host_str, e.what()));
}
@@ -41,7 +42,7 @@ void set_hinted_handoff(http_context& ctx, routes& r, sharded<service::storage_p
return hosts;
};

- std::vector<gms::inet_address> target_hosts = parse_hosts_list(req->get_query_param("target_hosts"));
+ std::vector<locator::host_id> target_hosts = parse_hosts_list(req->get_query_param("target_hosts"));
return proxy.local().create_hint_sync_point(std::move(target_hosts)).then([] (db::hints::sync_point sync_point) {
return json::json_return_type(sync_point.encode());
});
diff --git a/db/hints/manager.cc b/db/hints/manager.cc
index 95d5ea0f2a0..17332a0f52d 100644
--- a/db/hints/manager.cc
+++ b/db/hints/manager.cc
@@ -266,21 +266,14 @@ void manager::forbid_hints_for_eps_with_pending_hints() {
}
}

-sync_point::shard_rps manager::calculate_current_sync_point(std::span<const gms::inet_address> target_eps) const {
+sync_point::shard_rps manager::calculate_current_sync_point(std::span<const locator::host_id> target_eps) const {
sync_point::shard_rps rps;
- const auto tmptr = _proxy.get_token_metadata_ptr();

for (auto addr : target_eps) {
- const auto hid = tmptr->get_host_id_if_known(addr);
- // Ignore the IPs that we cannot map.
- if (!hid) {
- continue;
- }
-
- auto it = _ep_managers.find(*hid);
+ auto it = _ep_managers.find(addr);
if (it != _ep_managers.end()) {
const hint_endpoint_manager& ep_man = it->second;
- rps[*hid] = ep_man.last_written_replay_position();
+ rps[addr] = ep_man.last_written_replay_position();
}
}

diff --git a/main.cc b/main.cc
index 7cd2bef73e4..164cdc47167 100644
--- a/main.cc
+++ b/main.cc
@@ -2234,7 +2234,7 @@ To start the scylla server proper, simply invoke as: scylla server (or just scyl
supervisor::notify("allow replaying hints");
proxy.invoke_on_all(&service::storage_proxy::allow_replaying_hints).get();

- api::set_hinted_handoff(ctx, proxy).get();
+ api::set_hinted_handoff(ctx, proxy, gossiper).get();
auto stop_hinted_handoff_api = defer_verbose_shutdown("hinted handoff API", [&ctx] {
api::unset_hinted_handoff(ctx).get();
});
diff --git a/repair/row_level.cc b/repair/row_level.cc
index cb8843faaa8..7ef7aca25a8 100644
--- a/repair/row_level.cc
+++ b/repair/row_level.cc
@@ -2317,7 +2317,7 @@ future<repair_flush_hints_batchlog_response> repair_service::repair_flush_hints_
auto flush_time = now;
if (cache_disabled || (now - _flush_hints_batchlog_time > cache_time)) {
// Empty targets meants all nodes
- db::hints::sync_point sync_point = co_await _sp.local().create_hint_sync_point(std::vector<gms::inet_address>{});
+ db::hints::sync_point sync_point = co_await _sp.local().create_hint_sync_point(std::vector<locator::host_id>{});
lowres_clock::time_point deadline = lowres_clock::now() + req.hints_timeout;
try {
bool bm_throw = utils::get_local_injector().enter("repair_flush_hints_batchlog_handler_bm_uninitialized");
diff --git a/service/storage_proxy.cc b/service/storage_proxy.cc
index 6265191f121..7776c7fd988 100644
--- a/service/storage_proxy.cc
+++ b/service/storage_proxy.cc
@@ -6796,7 +6796,7 @@ const db::hints::host_filter& storage_proxy::get_hints_host_filter() const {
return _hints_manager.get_host_filter();
}

-future<db::hints::sync_point> storage_proxy::create_hint_sync_point(std::vector<gms::inet_address> target_hosts) const {
+future<db::hints::sync_point> storage_proxy::create_hint_sync_point(std::vector<locator::host_id> target_hosts) const {
db::hints::sync_point spoint;
spoint.regular_per_shard_rps.resize(smp::count);
spoint.mv_per_shard_rps.resize(smp::count);
--
2.47.1

[Kamil Braun]

typo in commit title.

Piotr or Dawid, please review

Maybe we should create a version of this API which accepts host IDs and
deprecate this one.

no longer an "addr"