Skip to content

feat(cluster): Implement CLUSTER SLOTS. #1289

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
merged 2 commits into from
May 28, 2023
Merged

feat(cluster): Implement CLUSTER SLOTS. #1289

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
9ff1bc5
feat(cluster): Implement `CLUSTER SLOTS`.
May 24, 2023
3181c0b
Remove `-` error message prefix
May 28, 2023
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
94 changes: 40 additions & 54 deletions src/server/cluster/cluster_family.cc
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -35,15 +35,7 @@ using namespace std;
using namespace facade;
using CI = CommandId;

void BuildClusterSlotNetworkInfo(ConnectionContext* cntx, std::string_view host, uint32_t port,
std::string_view id) {
constexpr unsigned int kNetworkInfoSize = 3;

(*cntx)->StartArray(kNetworkInfoSize);
(*cntx)->SendBulkString(host);
(*cntx)->SendLong(port);
(*cntx)->SendBulkString(id);
}
constexpr string_view kClusterNotConfigured = "-Cluster is not yet configured";
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

why did you add the '-' char at the beginning?

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I should not have, sorry!


} // namespace

Expand Down Expand Up @@ -200,60 +192,54 @@ void ClusterShardsImpl(const ClusterConfig::ClusterShards& config, ConnectionCon

void ClusterFamily::ClusterShards(ConnectionContext* cntx) {
if (is_emulated_cluster_) {
ClusterConfig::ClusterShards config{GetEmulatedShardInfo(cntx)};
return ClusterShardsImpl(config, cntx);
return ClusterShardsImpl({GetEmulatedShardInfo(cntx)}, cntx);
} else if (cluster_config_->IsConfigured()) {
return ClusterShardsImpl(cluster_config_->GetConfig(), cntx);
} else {
return (*cntx)->SendError("Cluster is not yet configured");
return (*cntx)->SendError(kClusterNotConfigured);
}
}

void ClusterFamily::ClusterSlots(ConnectionContext* cntx) {
namespace {
void ClusterSlotsImpl(const ClusterConfig::ClusterShards& config, ConnectionContext* cntx) {
// For more details https://redis.io/commands/cluster-slots/
constexpr unsigned int kClustersShardingCount = 1;
constexpr unsigned int kNoReplicaInfoSize = 3;
constexpr unsigned int kWithReplicaInfoSize = 4;

/* Format: 1) 1) start slot
* 2) end slot
* 3) 1) master IP
* 2) master port
* 3) node ID
* 4) 1) replica IP (optional)
* 2) replica port
* 3) node ID
* ... note that in this case, only 1 slot
*/
ServerState& etl = *ServerState::tlocal();
// we have 3 cases here
// 1. This is a stand alone, in this case we only sending local information
// 2. We are the master, and we have replica, in this case send us as master
// 3. We are replica to a master, sends the information about us as replica
(*cntx)->StartArray(kClustersShardingCount);
if (etl.is_master) {
std::string cluster_announce_ip = absl::GetFlag(FLAGS_cluster_announce_ip);
std::string preferred_endpoint =
cluster_announce_ip.empty() ? cntx->owner()->LocalBindAddress() : cluster_announce_ip;
auto vec = server_family_->GetDflyCmd()->GetReplicasRoleInfo();
unsigned int info_len = vec.empty() ? kNoReplicaInfoSize : kWithReplicaInfoSize;
(*cntx)->StartArray(info_len);
(*cntx)->SendLong(0); // start sharding range
(*cntx)->SendLong(ClusterConfig::kMaxSlotNum); // end sharding range
BuildClusterSlotNetworkInfo(cntx, preferred_endpoint, absl::GetFlag(FLAGS_port),
server_family_->master_id());
if (!vec.empty()) { // info about the replica
const auto& info = vec[0];
BuildClusterSlotNetworkInfo(cntx, info.address, info.listening_port, etl.remote_client_id_);
auto WriteNode = [&](const ClusterConfig::Node& node) {
constexpr unsigned int kNodeSize = 3;
(*cntx)->StartArray(kNodeSize);
(*cntx)->SendBulkString(node.ip);
(*cntx)->SendLong(node.port);
(*cntx)->SendBulkString(node.id);
};

unsigned int slot_ranges = 0;
for (const auto& shard : config) {
slot_ranges += shard.slot_ranges.size();
}

(*cntx)->StartArray(slot_ranges);
for (const auto& shard : config) {
for (const auto& slot_range : shard.slot_ranges) {
const unsigned int array_size =
/* slot-start, slot-end */ 2 + /* master */ 1 + /* replicas */ shard.replicas.size();
(*cntx)->StartArray(array_size);
(*cntx)->SendLong(slot_range.start);
(*cntx)->SendLong(slot_range.end);
WriteNode(shard.master);
for (const auto& replica : shard.replicas) {
WriteNode(replica);
}
}
}
}
} // namespace

void ClusterFamily::ClusterSlots(ConnectionContext* cntx) {
if (is_emulated_cluster_) {
return ClusterSlotsImpl({GetEmulatedShardInfo(cntx)}, cntx);
} else if (cluster_config_->IsConfigured()) {
return ClusterSlotsImpl(cluster_config_->GetConfig(), cntx);
} else {
Replica::Info info = server_family_->GetReplicaInfo();
(*cntx)->StartArray(kWithReplicaInfoSize);
(*cntx)->SendLong(0); // start sharding range
(*cntx)->SendLong(ClusterConfig::kMaxSlotNum); // end sharding range
BuildClusterSlotNetworkInfo(cntx, info.host, info.port, server_family_->GetReplicaMasterId());
BuildClusterSlotNetworkInfo(cntx, cntx->owner()->LocalBindAddress(), absl::GetFlag(FLAGS_port),
server_family_->master_id());
return (*cntx)->SendError(kClusterNotConfigured);
}
}

Expand Down
59 changes: 52 additions & 7 deletions src/server/cluster_family_test.cc
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -66,6 +66,7 @@ TEST_F(ClusterFamilyTest, ClusterConfigInvalidJSON) {
EXPECT_THAT(cluster_info, HasSubstr("cluster_size:0"));

EXPECT_THAT(Run({"cluster", "shards"}), ErrArg("Cluster is not yet configured"));
EXPECT_THAT(Run({"cluster", "slots"}), ErrArg("Cluster is not yet configured"));
}

TEST_F(ClusterFamilyTest, ClusterConfigInvalidConfig) {
Expand Down Expand Up @@ -193,8 +194,13 @@ TEST_F(ClusterFamilyTest, ClusterConfigNoReplicas) {
EXPECT_THAT(Run({"get", "x"}).GetString(),
testing::MatchesRegex(R"(MOVED [0-9]+ 10.0.0.1:7000)"));

// TODO: Use "CLUSTER SLOTS" and "CLUSTER SHARDS" once implemented to verify new configuration
// takes effect.
EXPECT_THAT(Run({"cluster", "slots"}),
RespArray(ElementsAre(IntArg(0), //
IntArg(16'383), //
RespArray(ElementsAre( //
"10.0.0.1", //
IntArg(7'000), //
"abcd1234")))));
}

TEST_F(ClusterFamilyTest, ClusterConfigFull) {
Expand Down Expand Up @@ -252,8 +258,17 @@ TEST_F(ClusterFamilyTest, ClusterConfigFull) {
"replication-offset", IntArg(0), //
"health", "online")))))));

// TODO: Use "CLUSTER SLOTS" and "CLUSTER SHARDS" once implemented to verify new configuration
// takes effect.
EXPECT_THAT(Run({"cluster", "slots"}),
RespArray(ElementsAre(IntArg(0), //
IntArg(16'383), //
RespArray(ElementsAre( //
"10.0.0.1", //
IntArg(7'000), //
"abcd1234")), //
RespArray(ElementsAre( //
"10.0.0.10", //
IntArg(8'000), //
"wxyz")))));
}

TEST_F(ClusterFamilyTest, ClusterConfigFullMultipleInstances) {
Expand Down Expand Up @@ -352,6 +367,29 @@ TEST_F(ClusterFamilyTest, ClusterConfigFullMultipleInstances) {
"replication-offset", IntArg(0), //
"health", "online")))))))));

EXPECT_THAT(Run({"cluster", "slots"}),
RespArray(ElementsAre( //
RespArray(ElementsAre(IntArg(0), //
IntArg(10'000), //
RespArray(ElementsAre( //
"10.0.0.1", //
IntArg(7'000), //
"abcd1234")), //
RespArray(ElementsAre( //
"10.0.0.10", //
IntArg(8'000), //
"wxyz")))), //
RespArray(ElementsAre(IntArg(10'001), //
IntArg(16'383), //
RespArray(ElementsAre( //
"10.0.0.2", //
IntArg(7'001), //
"efgh7890")), //
RespArray(ElementsAre( //
"10.0.0.11", //
IntArg(8'001), //
"qwerty")))))));

absl::InsecureBitGen eng;
while (true) {
string random_key = GetRandomHex(eng, 40);
Expand All @@ -376,9 +414,6 @@ TEST_F(ClusterFamilyTest, ClusterConfigFullMultipleInstances) {
testing::MatchesRegex(R"(MOVED [0-9]+ 10.0.0.2:7001)"));
break;
}

// TODO: Use "CLUSTER SLOTS" and "CLUSTER SHARDS" once implemented to verify new configuration
// takes effect.
}

TEST_F(ClusterFamilyTest, ClusterGetSlotInfo) {
Expand Down Expand Up @@ -491,5 +526,15 @@ TEST_F(ClusterFamilyEmulatedTest, ClusterShards) {
"health", "online")))))));
}

TEST_F(ClusterFamilyEmulatedTest, ClusterSlots) {
EXPECT_THAT(Run({"cluster", "slots"}),
RespArray(ElementsAre(IntArg(0), //
IntArg(16383), //
RespArray(ElementsAre( //
"fake-host", //
IntArg(6379), //
RunAdmin({"dflycluster", "myid"}).GetString())))));
}

} // namespace
} // namespace dfly