apple/foundationdb
1
0
Fork 0
mirror of https://github.com/apple/foundationdb.git synced 2025-05-15 10:22:20 +08:00
Code Issues Packages Projects Releases Wiki Activity

add knob enabling cross cluster support (default true)

Browse Source
This commit is contained in:
Dan Lambright 2021-04-05 15:52:48 -04:00
parent 065c4fdd5a
commit 60d27d05d8
6 changed files with 57 additions and 42 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

81
fdbserver/Coordination.actor.cpp
View File

@ -568,17 +568,18 @@ ACTOR Future<Void> leaderServer(LeaderElectionRegInterface interf, OnDemandStore
info.forward = forward.get().serializedInfo; info.forward = forward.get().serializedInfo;
req.reply.send(CachedSerialization<ClientDBInfo>(info)); req.reply.send(CachedSerialization<ClientDBInfo>(info));
} else { } else {
StringRef reqClusterName = getClusterName(req.clusterKey); StringRef clusterName = ccf->getConnectionString().clusterKeyName();
StringRef clusterName = getClusterName(ccf->getConnectionString().clusterKey()); if (! SERVER_KNOBS->ENABLE_CROSS_CLUSTER_SUPPORT &&
if (reqClusterName.compare(clusterName) || getClusterName(req.clusterKey).compare(clusterName)) {
ccf->getConnectionString().coordinators() != req.coordinators) { TraceEvent(SevError, "CCFMismatch")
TraceEvent(SevWarnAlways, "CCFMismatch")
.detail("RequestType", "OpenDatabaseCoordRequest") .detail("RequestType", "OpenDatabaseCoordRequest")
.detail("LocalCS", ccf->getConnectionString().toString()) .detail("LocalCS", ccf->getConnectionString().toString())
.detail("IncomingClusterKey", req.clusterKey) .detail("IncomingClusterKey", req.clusterKey)
.detail("IncomingCoordinators", describeList(req.coordinators, req.coordinators.size())); .detail("IncomingCoordinators", describeList(req.coordinators, req.coordinators.size()));
req.reply.sendError(wrong_connection_file());
} else {
regs.getInterface(req.clusterKey, id).openDatabase.send(req);
} }
regs.getInterface(req.clusterKey, id).openDatabase.send(req);
} }
} }
when(ElectionResultRequest req = waitNext(interf.electionResult.getFuture())) { when(ElectionResultRequest req = waitNext(interf.electionResult.getFuture())) {
@ -586,17 +587,19 @@ ACTOR Future<Void> leaderServer(LeaderElectionRegInterface interf, OnDemandStore
if (forward.present()) { if (forward.present()) {
req.reply.send(forward.get()); req.reply.send(forward.get());
} else { } else {
StringRef reqClusterName = getClusterName(req.key); StringRef clusterName = ccf->getConnectionString().clusterKeyName();
StringRef clusterName = getClusterName(ccf->getConnectionString().clusterKey()); if (! SERVER_KNOBS->ENABLE_CROSS_CLUSTER_SUPPORT &&
if (reqClusterName.compare(clusterName) || getClusterName(req.key).compare(clusterName)) {
ccf->getConnectionString().coordinators() != req.coordinators) { TraceEvent(SevError, "CCFMismatch")
TraceEvent(SevWarnAlways, "CCFMismatch")
.detail("RequestType", "ElectionResultRequest") .detail("RequestType", "ElectionResultRequest")
.detail("LocalCS", ccf->getConnectionString().toString()) .detail("LocalCS", ccf->getConnectionString().toString())
.detail("IncomingClusterKey", req.key) .detail("IncomingClusterKey", req.key)
.detail("ClusterKey", ccf->getConnectionString().clusterKey())
.detail("IncomingCoordinators", describeList(req.coordinators, req.coordinators.size())); .detail("IncomingCoordinators", describeList(req.coordinators, req.coordinators.size()));
req.reply.sendError(wrong_connection_file());
} else {
regs.getInterface(req.key, id).electionResult.send(req);
} }
regs.getInterface(req.key, id).electionResult.send(req);
} }
} }
when(GetLeaderRequest req = waitNext(interf.getLeader.getFuture())) { when(GetLeaderRequest req = waitNext(interf.getLeader.getFuture())) {
@ -604,16 +607,18 @@ ACTOR Future<Void> leaderServer(LeaderElectionRegInterface interf, OnDemandStore
if (forward.present()) if (forward.present())
req.reply.send(forward.get()); req.reply.send(forward.get());
else { else {
StringRef reqClusterName = getClusterName(req.key); StringRef clusterName = ccf->getConnectionString().clusterKeyName();
StringRef clusterName = getClusterName(ccf->getConnectionString().clusterKey()); if (! SERVER_KNOBS->ENABLE_CROSS_CLUSTER_SUPPORT &&
if (reqClusterName.compare(clusterName)) { getClusterName(req.key).compare(clusterName)) {
TraceEvent(SevWarnAlways, "CCFMismatch") TraceEvent(SevError, "CCFMismatch")
.detail("RequestType", "GetLeaderRequest") .detail("RequestType", "GetLeaderRequest")
.detail("LocalCS", ccf->getConnectionString().toString()) .detail("LocalCS", ccf->getConnectionString().toString())
.detail("IncomingClusterKey", req.key) .detail("IncomingClusterKey", req.key)
.detail("Key", reqClusterName).detail("Key2",clusterName); .detail("ClusterKey", ccf->getConnectionString().clusterKey());
req.reply.sendError(wrong_connection_file());
} else {
regs.getInterface(req.key, id).getLeader.send(req);
} }
regs.getInterface(req.key, id).getLeader.send(req);
} }
} }
when(CandidacyRequest req = waitNext(interf.candidacy.getFuture())) { when(CandidacyRequest req = waitNext(interf.candidacy.getFuture())) {
@ -621,15 +626,17 @@ ACTOR Future<Void> leaderServer(LeaderElectionRegInterface interf, OnDemandStore
if (forward.present()) if (forward.present())
req.reply.send(forward.get()); req.reply.send(forward.get());
else { else {
StringRef reqClusterName = getClusterName(req.key); StringRef clusterName = ccf->getConnectionString().clusterKeyName();
StringRef clusterName = getClusterName(ccf->getConnectionString().clusterKey()); if (! SERVER_KNOBS->ENABLE_CROSS_CLUSTER_SUPPORT &&
if (reqClusterName.compare(clusterName)) { getClusterName(req.key).compare(clusterName)) {
TraceEvent(SevWarnAlways, "CCFMismatch") TraceEvent(SevError, "CCFMismatch")
.detail("RequestType", "CandidacyRequest") .detail("RequestType", "CandidacyRequest")
.detail("LocalCS", ccf->getConnectionString().toString()) .detail("LocalCS", ccf->getConnectionString().toString())
.detail("IncomingClusterKey", req.key); .detail("IncomingClusterKey", req.key);
req.reply.sendError(wrong_connection_file());
} else {
regs.getInterface(req.key, id).candidacy.send(req);
} }
regs.getInterface(req.key, id).candidacy.send(req);
} }
} }
when(LeaderHeartbeatRequest req = waitNext(interf.leaderHeartbeat.getFuture())) { when(LeaderHeartbeatRequest req = waitNext(interf.leaderHeartbeat.getFuture())) {
@ -637,15 +644,17 @@ ACTOR Future<Void> leaderServer(LeaderElectionRegInterface interf, OnDemandStore
if (forward.present()) if (forward.present())
req.reply.send(LeaderHeartbeatReply{ false }); req.reply.send(LeaderHeartbeatReply{ false });
else { else {
StringRef reqClusterName = getClusterName(req.key); StringRef clusterName = ccf->getConnectionString().clusterKeyName();
StringRef clusterName = getClusterName(ccf->getConnectionString().clusterKey()); if (! SERVER_KNOBS->ENABLE_CROSS_CLUSTER_SUPPORT &&
if (reqClusterName.compare(clusterName)) { getClusterName(req.key).compare(clusterName)) {
TraceEvent(SevWarnAlways, "CCFMismatch") TraceEvent(SevError, "CCFMismatch")
.detail("RequestType", "LeaderHeartbeatRequest") .detail("RequestType", "LeaderHeartbeatRequest")
.detail("LocalCS", ccf->getConnectionString().toString()) .detail("LocalCS", ccf->getConnectionString().toString())
.detail("IncomingClusterKey", req.key); .detail("IncomingClusterKey", req.key);
req.reply.sendError(wrong_connection_file());
} else {
regs.getInterface(req.key, id).leaderHeartbeat.send(req);
} }
regs.getInterface(req.key, id).leaderHeartbeat.send(req);
} }
} }
when(ForwardRequest req = waitNext(interf.forward.getFuture())) { when(ForwardRequest req = waitNext(interf.forward.getFuture())) {
@ -653,17 +662,19 @@ ACTOR Future<Void> leaderServer(LeaderElectionRegInterface interf, OnDemandStore
if (forward.present()) if (forward.present())
req.reply.send(Void()); req.reply.send(Void());
else { else {
StringRef reqClusterName = getClusterName(req.key); StringRef clusterName = ccf->getConnectionString().clusterKeyName();
StringRef clusterName = getClusterName(ccf->getConnectionString().clusterKey()); if (! SERVER_KNOBS->ENABLE_CROSS_CLUSTER_SUPPORT &&
if (reqClusterName.compare(clusterName)) { getClusterName(req.key).compare(clusterName)) {
TraceEvent(SevWarnAlways, "CCFMismatch") TraceEvent(SevError, "CCFMismatch")
.detail("RequestType", "ForwardRequest") .detail("RequestType", "ForwardRequest")
.detail("LocalCS", ccf->getConnectionString().toString()) .detail("LocalCS", ccf->getConnectionString().toString())
.detail("IncomingClusterKey", req.key); .detail("IncomingClusterKey", req.key);
req.reply.sendError(wrong_connection_file());
} else {
forwarders.add(
LeaderRegisterCollection::setForward(&regs, req.key, ClusterConnectionString(req.conn.toString())));
regs.getInterface(req.key, id).forward.send(req);
} }
forwarders.add(LeaderRegisterCollection::setForward(&regs, req.key,
ClusterConnectionString(req.conn.toString())));
regs.getInterface(req.key, id).forward.send(req);
} }
} }
when(wait(forwarders.getResult())) { when(wait(forwarders.getResult())) {

1
fdbserver/Knobs.cpp
View File

@ -631,7 +631,6 @@ void ServerKnobs::initialize(bool randomize, ClientKnobs* clientKnobs, bool isSi
// Coordination // Coordination
init( COORDINATED_STATE_ONCONFLICT_POLL_INTERVAL, 1.0 ); if( randomize && BUGGIFY ) COORDINATED_STATE_ONCONFLICT_POLL_INTERVAL = 10.0; init( COORDINATED_STATE_ONCONFLICT_POLL_INTERVAL, 1.0 ); if( randomize && BUGGIFY ) COORDINATED_STATE_ONCONFLICT_POLL_INTERVAL = 10.0;
init( FORWARD_REQUEST_TOO_OLD, 600.0 ); if( randomize && BUGGIFY ) FORWARD_REQUEST_TOO_OLD = 60.0;
init( ENABLE_CROSS_CLUSTER_SUPPORT, true ); if( randomize && BUGGIFY ) ENABLE_CROSS_CLUSTER_SUPPORT = false; init( ENABLE_CROSS_CLUSTER_SUPPORT, true ); if( randomize && BUGGIFY ) ENABLE_CROSS_CLUSTER_SUPPORT = false;
// Buggification // Buggification

4
fdbserver/Knobs.h
View File

@ -559,9 +559,7 @@ public:
// Coordination // Coordination
double COORDINATED_STATE_ONCONFLICT_POLL_INTERVAL; double COORDINATED_STATE_ONCONFLICT_POLL_INTERVAL;
double FORWARD_REQUEST_TOO_OLD; bool ENABLE_CROSS_CLUSTER_SUPPORT; // Allow a coordinator to serve requests whose connection string does not match the local descriptor
bool ENABLE_CROSS_CLUSTER_SUPPORT; // Allow a coordinator to serve requests whose connection string does not match
// the local copy
// Buggification // Buggification
double BUGGIFIED_EVENTUAL_CONSISTENCY; double BUGGIFIED_EVENTUAL_CONSISTENCY;

2
fdbserver/SimulatedCluster.actor.cpp
View File

@ -1691,7 +1691,7 @@ void setupSimulatedSystem(vector<Future<Void>>* systemActors,
} else if (testConfig.extraDB == 2) { } else if (testConfig.extraDB == 2) {
// The DR database is a new database // The DR database is a new database
g_simulator.extraDB = g_simulator.extraDB =
new ClusterConnectionString(extraCoordinatorAddresses, LiteralStringRef("ExtraCluster:0")); new ClusterConnectionString(extraCoordinatorAddresses, LiteralStringRef("ExtraCluster:0"));
} else if (testConfig.extraDB == 3) { } else if (testConfig.extraDB == 3) {
// The DR database is the same database // The DR database is the same database
g_simulator.extraDB = new ClusterConnectionString(coordinatorAddresses, LiteralStringRef("TestCluster:0")); g_simulator.extraDB = new ClusterConnectionString(coordinatorAddresses, LiteralStringRef("TestCluster:0"));

8
fdbserver/workloads/ConfigureDatabase.actor.cpp
View File

@ -270,6 +270,7 @@ struct ConfigureDatabaseWorkload : TestWorkload {
return Void(); return Void();
} }
state int randomChoice = deterministicRandom()->randomInt(0, 8); state int randomChoice = deterministicRandom()->randomInt(0, 8);
if (randomChoice == 0) { if (randomChoice == 0) {
wait(success( wait(success(
runRYWTransaction(cx, [=](Reference<ReadYourWritesTransaction> tr) -> Future<Optional<Value>> { runRYWTransaction(cx, [=](Reference<ReadYourWritesTransaction> tr) -> Future<Optional<Value>> {
@ -316,8 +317,13 @@ struct ConfigureDatabaseWorkload : TestWorkload {
} else if (randomChoice == 4) { } else if (randomChoice == 4) {
//TraceEvent("ConfigureTestQuorumBegin").detail("NewQuorum", s); //TraceEvent("ConfigureTestQuorumBegin").detail("NewQuorum", s);
auto ch = autoQuorumChange(); auto ch = autoQuorumChange();
std::string desiredClusterName = "NewName%d";
if (! SERVER_KNOBS->ENABLE_CROSS_CLUSTER_SUPPORT) {
// if configuration does not allow changing the descriptor, pass empty string (keep old descriptor)
desiredClusterName = "";
}
if (deterministicRandom()->randomInt(0, 2)) if (deterministicRandom()->randomInt(0, 2))
ch = nameQuorumChange(format("NewName%d", deterministicRandom()->randomInt(0, 100)), ch); ch = nameQuorumChange(format(desiredClusterName.c_str(), deterministicRandom()->randomInt(0, 100)), ch);
wait(success(changeQuorum(cx, ch))); wait(success(changeQuorum(cx, ch)));
//TraceEvent("ConfigureTestConfigureEnd").detail("NewQuorum", s); //TraceEvent("ConfigureTestConfigureEnd").detail("NewQuorum", s);
} else if (randomChoice == 5) { } else if (randomChoice == 5) {

3
fdbserver/workloads/SpecialKeySpaceCorrectness.actor.cpp
View File

@ -936,7 +936,8 @@ struct SpecialKeySpaceCorrectnessWorkload : TestWorkload {
// test change coordinators and cluster description // test change coordinators and cluster description
// we randomly pick one process(not coordinator) and add it, in this case, it should always succeed // we randomly pick one process(not coordinator) and add it, in this case, it should always succeed
{ {
state std::string new_cluster_description = deterministicRandom()->randomAlphaNumeric(8); // choose a new description if configuration allows transactions across differently named clusters
state std::string new_cluster_description = SERVER_KNOBS->ENABLE_CROSS_CLUSTER_SUPPORT ? deterministicRandom()->randomAlphaNumeric(8) : cs.clusterKeyName().toString();
state std::string new_coordinator_process; state std::string new_coordinator_process;
state std::vector<std::string> old_coordinators_processes; state std::vector<std::string> old_coordinators_processes;
state bool possible_to_add_coordinator; state bool possible_to_add_coordinator;