Hawkin
/
arangodb
mirror of https://gitee.com/bigwinds/arangodb
1
0
Fork 0
Code Issues Projects Releases Wiki Activity

Bug fix/no support for inquiry in send transaction with failover less verbosity (#3847)

This commit is contained in:
Kaveh Vahedipour 2017-11-30 14:54:22 +01:00 committed by Jan
parent e3277a493c
commit 11cfa74495
5 changed files with 74 additions and 214 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

9
arangod/Agency/AgencyComm.cpp
View File

@ -1382,6 +1382,8 @@ AgencyCommResult AgencyComm::sendWithFailover(
bool isInquiry = false; // Set to true whilst we investigate a potentially
// failed transaction.
static std::string const writeURL {"/_api/agency/write"};
bool isWriteTrans = (initialUrl == writeURL);
while (true) { // will be left by timeout eventually
// If for some reason we did not find an agency endpoint, we bail out:
@ -1467,7 +1469,8 @@ AgencyCommResult AgencyComm::sendWithFailover(
// the operation. If it actually was done, we are good. If not, we
// can retry. If in doubt, we have to retry inquire until the global
// timeout is reached.
if (!clientIds.empty() && result._sent &&
// Also note that we only inquire about WriteTransactions.
if (isWriteTrans && !clientIds.empty() && result._sent &&
(result._statusCode == 0 || result._statusCode == 503)) {
isInquiry = true;
}
@ -1510,11 +1513,7 @@ AgencyCommResult AgencyComm::sendWithFailover(
if (outer.isObject() && outer.hasKey("results")) {
VPackSlice results = outer.get("results");
if (results.length() > 0) {
#ifdef ARANGODB_ENABLE_MAINTAINER_MODE
LOG_TOPIC(WARN, Logger::AGENCYCOMM)
#else
LOG_TOPIC(DEBUG, Logger::AGENCYCOMM)
#endif
<< "Inquired " << resultBody->toJson();
AgencyCommManager::MANAGER->release(std::move(connection), endpoint);
break;

230
arangod/Agency/RestAgencyHandler.cpp
View File

@ -71,6 +71,16 @@ inline RestStatus RestAgencyHandler::reportUnknownMethod() {
return RestStatus::DONE;
}
inline RestStatus RestAgencyHandler::reportMessage(
rest::ResponseCode code, std::string const& message) {
LOG_TOPIC(DEBUG, Logger::AGENCY) << message;
Builder body;
{ VPackObjectBuilder b(&body);
body.add("message", VPackValue(message)); }
generateResult(code, body.slice());
return RestStatus::DONE;
}
void RestAgencyHandler::redirectRequest(std::string const& leaderId) {
try {
std::string url = Endpoint::uriForm(_agent->config().poolAt(leaderId)) +
@ -79,13 +89,7 @@ void RestAgencyHandler::redirectRequest(std::string const& leaderId) {
_response->setHeaderNC(StaticStrings::Location, url);
LOG_TOPIC(DEBUG, Logger::AGENCY) << "Sending 307 redirect to " << url;
} catch (std::exception const& e) {
Builder body;
body.openObject();
body.add("message", VPackValue("No leader"));
body.close();
generateResult(rest::ResponseCode::SERVICE_UNAVAILABLE, body.slice());
LOG_TOPIC(DEBUG, Logger::AGENCY) << "We don't know who the leader is, "
"caught exception in redirectRequest: " << e.what();
reportMessage(rest::ResponseCode::SERVICE_UNAVAILABLE, "No leader");
}
}
@ -100,46 +104,23 @@ RestStatus RestAgencyHandler::handleTransient() {
try {
query = _request->toVelocyPackBuilderPtr();
} catch (std::exception const& e) {
LOG_TOPIC(ERR, Logger::AGENCY)
<< e.what() << " " << __FILE__ << ":" << __LINE__;
Builder body;
body.openObject();
body.add("message", VPackValue(e.what()));
body.close();
generateResult(rest::ResponseCode::BAD, body.slice());
return RestStatus::DONE;
return reportMessage(rest::ResponseCode::BAD, e.what());
}
// Need Array input
if (!query->slice().isArray()) {
Builder body;
body.openObject();
body.add(
"message", VPackValue("Expecting array of arrays as body for writes"));
body.close();
generateResult(rest::ResponseCode::BAD, body.slice());
return RestStatus::DONE;
return reportMessage(
rest::ResponseCode::BAD, "Expecting array of arrays as body for writes");
}
// Empty request array
if (query->slice().length() == 0) {
Builder body;
body.openObject();
body.add("message", VPackValue("Empty request."));
body.close();
generateResult(rest::ResponseCode::BAD, body.slice());
return RestStatus::DONE;
return reportMessage(rest::ResponseCode::BAD, "Empty request");
}
// Leadership established?
if (_agent->size() > 1 && _agent->leaderID() == NO_LEADER) {
Builder body;
body.openObject();
body.add("message", VPackValue("No leader"));
body.close();
generateResult(rest::ResponseCode::SERVICE_UNAVAILABLE, body.slice());
LOG_TOPIC(DEBUG, Logger::AGENCY) << "We don't know who the leader is";
return RestStatus::DONE;
return reportMessage(rest::ResponseCode::SERVICE_UNAVAILABLE, "No leader");
}
trans_ret_t ret;
@ -147,37 +128,24 @@ RestStatus RestAgencyHandler::handleTransient() {
try {
ret = _agent->transient(query);
} catch (std::exception const& e) {
Builder body;
body.openObject();
body.add("message", VPackValue(e.what()));
body.close();
generateResult(rest::ResponseCode::BAD, body.slice());
return RestStatus::DONE;
return reportMessage(rest::ResponseCode::BAD, e.what());
}
// We're leading and handling the request
if (ret.accepted) {
generateResult(
(ret.failed==0) ?
rest::ResponseCode::OK : rest::ResponseCode::PRECONDITION_FAILED,
ret.result->slice());
} else { // Redirect to leader
if (_agent->leaderID() == NO_LEADER) {
Builder body;
body.openObject();
body.add("message", VPackValue("No leader"));
body.close();
generateResult(rest::ResponseCode::SERVICE_UNAVAILABLE, body.slice());
LOG_TOPIC(DEBUG, Logger::AGENCY) << "We don't know who the leader is";
return RestStatus::DONE;
return reportMessage(rest::ResponseCode::SERVICE_UNAVAILABLE, "No leader");
} else {
TRI_ASSERT(ret.redirect != _agent->id());
redirectRequest(ret.redirect);
}
}
return RestStatus::DONE;
}
@ -251,46 +219,23 @@ RestStatus RestAgencyHandler::handleWrite() {
try {
query = _request->toVelocyPackBuilderPtr();
} catch (std::exception const& e) {
LOG_TOPIC(ERR, Logger::AGENCY)
<< e.what() << " " << __FILE__ << ":" << __LINE__;
Builder body;
body.openObject();
body.add("message", VPackValue(e.what()));
body.close();
generateResult(rest::ResponseCode::BAD, body.slice());
return RestStatus::DONE;
return reportMessage(rest::ResponseCode::BAD, e.what());
}
// Need Array input
if (!query->slice().isArray()) {
Builder body;
body.openObject();
body.add(
"message", VPackValue("Expecting array of arrays as body for writes"));
body.close();
generateResult(rest::ResponseCode::BAD, body.slice());
return RestStatus::DONE;
return reportMessage(
rest::ResponseCode::BAD, "Expecting array of arrays as body for writes");
}
// Empty request array
if (query->slice().length() == 0) {
Builder body;
body.openObject();
body.add("message", VPackValue("Empty request."));
body.close();
generateResult(rest::ResponseCode::BAD, body.slice());
return RestStatus::DONE;
return reportMessage(rest::ResponseCode::BAD, "Empty request.");
}
// Leadership established?
if (_agent->size() > 1 && _agent->leaderID() == NO_LEADER) {
Builder body;
body.openObject();
body.add("message", VPackValue("No leader"));
body.close();
generateResult(rest::ResponseCode::SERVICE_UNAVAILABLE, body.slice());
LOG_TOPIC(DEBUG, Logger::AGENCY) << "We don't know who the leader is";
return RestStatus::DONE;
return reportMessage(rest::ResponseCode::SERVICE_UNAVAILABLE, "No leader");
}
// Do write
@ -298,25 +243,14 @@ RestStatus RestAgencyHandler::handleWrite() {
try {
ret = _agent->write(query);
} catch (std::exception const& e) {
LOG_TOPIC(DEBUG, Logger::AGENCY) << "Malformed write query " << query;
Builder body;
body.openObject();
body.add("message",
VPackValue(std::string("Malformed write query") + e.what()));
body.close();
generateResult(rest::ResponseCode::BAD, body.slice());
return RestStatus::DONE;
return reportMessage(rest::ResponseCode::BAD, e.what());
}
// We're leading and handling the request
if (ret.accepted) {
bool found;
std::string call_mode = _request->header("x-arangodb-agency-mode", found);
if (!found) {
call_mode = "waitForCommitted";
}
if (!found) { call_mode = "waitForCommitted"; }
size_t errors = 0;
Builder body;
body.openObject();
@ -366,13 +300,7 @@ RestStatus RestAgencyHandler::handleWrite() {
} else { // Redirect to leader
if (_agent->leaderID() == NO_LEADER) {
Builder body;
body.openObject();
body.add("message", VPackValue("No leader"));
body.close();
generateResult(rest::ResponseCode::SERVICE_UNAVAILABLE, body.slice());
LOG_TOPIC(DEBUG, Logger::AGENCY) << "We don't know who the leader is";
return RestStatus::DONE;
return reportMessage(rest::ResponseCode::SERVICE_UNAVAILABLE, "No leader");
} else {
TRI_ASSERT(ret.redirect != _agent->id());
redirectRequest(ret.redirect);
@ -393,46 +321,23 @@ RestStatus RestAgencyHandler::handleTransact() {
try {
query = _request->toVelocyPackBuilderPtr();
} catch (std::exception const& e) {
LOG_TOPIC(ERR, Logger::AGENCY)
<< e.what() << " " << __FILE__ << ":" << __LINE__;
Builder body;
body.openObject();
body.add("message", VPackValue(e.what()));
body.close();
generateResult(rest::ResponseCode::BAD, body.slice());
return RestStatus::DONE;
return reportMessage(rest::ResponseCode::BAD, e.what());
}
// Need Array input
if (!query->slice().isArray()) {
Builder body;
body.openObject();
body.add(
"message", VPackValue("Expecting array of arrays as body for writes"));
body.close();
generateResult(rest::ResponseCode::BAD, body.slice());
return RestStatus::DONE;
return reportMessage(
rest::ResponseCode::BAD, "Expecting array of arrays as body for writes");
}
// Empty request array
if (query->slice().length() == 0) {
Builder body;
body.openObject();
body.add("message", VPackValue("Empty request."));
body.close();
generateResult(rest::ResponseCode::BAD, body.slice());
return RestStatus::DONE;
return reportMessage(rest::ResponseCode::BAD, "Empty request");
}
// Leadership established?
if (_agent->size() > 1 && _agent->leaderID() == NO_LEADER) {
Builder body;
body.openObject();
body.add("message", VPackValue("No leader"));
body.close();
generateResult(rest::ResponseCode::SERVICE_UNAVAILABLE, body.slice());
LOG_TOPIC(DEBUG, Logger::AGENCY) << "We don't know who the leader is";
return RestStatus::DONE;
return reportMessage(rest::ResponseCode::SERVICE_UNAVAILABLE, "No leader");
}
// Do write
@ -440,14 +345,7 @@ RestStatus RestAgencyHandler::handleTransact() {
try {
ret = _agent->transact(query);
} catch (std::exception const& e) {
LOG_TOPIC(DEBUG, Logger::AGENCY) << "Malformed write query " << query;
Builder body;
body.openObject();
body.add(
"message", VPackValue(std::string("Malformed write query") + e.what()));
body.close();
generateResult(rest::ResponseCode::BAD, body.slice());
return RestStatus::DONE;
return reportMessage(rest::ResponseCode::BAD, e.what());
}
// We're leading and handling the request
@ -464,13 +362,7 @@ RestStatus RestAgencyHandler::handleTransact() {
} else { // Redirect to leader
if (_agent->leaderID() == NO_LEADER) {
Builder body;
body.openObject();
body.add("message", VPackValue("No leader"));
body.close();
generateResult(rest::ResponseCode::SERVICE_UNAVAILABLE, body.slice());
LOG_TOPIC(DEBUG, Logger::AGENCY) << "We don't know who the leader is";
return RestStatus::DONE;
return reportMessage(rest::ResponseCode::SERVICE_UNAVAILABLE, "No leader");
} else {
TRI_ASSERT(ret.redirect != _agent->id());
redirectRequest(ret.redirect);
@ -499,27 +391,17 @@ RestStatus RestAgencyHandler::handleInquire() {
// Leadership established?
if (_agent->size() > 1 && _agent->leaderID() == NO_LEADER) {
Builder body;
body.openObject();
body.add("message", VPackValue("No leader"));
body.close();
generateResult(rest::ResponseCode::SERVICE_UNAVAILABLE, body.slice());
LOG_TOPIC(DEBUG, Logger::AGENCY) << "We don't know who the leader is";
return RestStatus::DONE;
return reportMessage(rest::ResponseCode::SERVICE_UNAVAILABLE, "No leader");
}
write_ret_t ret;
try {
ret = _agent->inquire(query);
} catch (std::exception const& e) {
generateError(
rest::ResponseCode::SERVER_ERROR, TRI_ERROR_INTERNAL, e.what());
return RestStatus::DONE;
return reportMessage(rest::ResponseCode::SERVER_ERROR, e.what());
}
if (ret.accepted) { // I am leading
Builder body;
bool failed = false;
{ VPackObjectBuilder b(&body);
@ -531,22 +413,16 @@ RestStatus RestAgencyHandler::handleInquire() {
for (auto const& index : ret.indices) {
body.add(VPackValue(index));
failed = (failed || index == 0);
}}}
}}
}
body.add("inquired", VPackValue(true));
}
generateResult(failed ? rest::ResponseCode::PRECONDITION_FAILED :
rest::ResponseCode::OK, body.slice());
} else { // Redirect to leader
if (_agent->leaderID() == NO_LEADER) {
Builder body;
body.openObject();
body.add("message", VPackValue("No leader"));
body.close();
generateResult(rest::ResponseCode::SERVICE_UNAVAILABLE, body.slice());
LOG_TOPIC(DEBUG, Logger::AGENCY) << "We don't know who the leader is";
return RestStatus::DONE;
return reportMessage(rest::ResponseCode::SERVICE_UNAVAILABLE, "No leader");
} else {
TRI_ASSERT(ret.redirect != _agent->id());
redirectRequest(ret.redirect);
@ -562,22 +438,13 @@ RestStatus RestAgencyHandler::handleRead() {
try {
query = _request->toVelocyPackBuilderPtr();
} catch (std::exception const& e) {
LOG_TOPIC(DEBUG, Logger::AGENCY)
<< e.what() << " " << __FILE__ << ":" << __LINE__;
generateError(rest::ResponseCode::BAD, 400);
return RestStatus::DONE;
return reportMessage(rest::ResponseCode::BAD, e.what());
}
if (_agent->size() > 1 && _agent->leaderID() == NO_LEADER) {
Builder body;
body.openObject();
body.add("message", VPackValue("No leader"));
body.close();
generateResult(rest::ResponseCode::SERVICE_UNAVAILABLE, body.slice());
LOG_TOPIC(DEBUG, Logger::AGENCY) << "We don't know who the leader is";
return RestStatus::DONE;
return reportMessage(rest::ResponseCode::SERVICE_UNAVAILABLE, "No leader");
}
read_ret_t ret = _agent->read(query);
if (ret.accepted) { // I am leading
@ -588,14 +455,7 @@ RestStatus RestAgencyHandler::handleRead() {
}
} else { // Redirect to leader
if (_agent->leaderID() == NO_LEADER) {
Builder body;
body.openObject();
body.add("message", VPackValue("No leader"));
body.close();
generateResult(rest::ResponseCode::SERVICE_UNAVAILABLE, body.slice());
LOG_TOPIC(DEBUG, Logger::AGENCY) << "We don't know who the leader is";
return RestStatus::DONE;
return reportMessage(rest::ResponseCode::SERVICE_UNAVAILABLE, "No leader");
} else {
TRI_ASSERT(ret.redirect != _agent->id());
redirectRequest(ret.redirect);

1
arangod/Agency/RestAgencyHandler.h
View File

@ -47,6 +47,7 @@ class RestAgencyHandler : public RestBaseHandler {
RestStatus reportErrorEmptyRequest();
RestStatus reportTooManySuffices();
RestStatus reportUnknownMethod();
RestStatus reportMessage(arangodb::rest::ResponseCode, std::string const&);
RestStatus handleStores();
RestStatus handleStore();
RestStatus handleRead();

11
arangod/Cluster/HeartbeatThread.cpp
View File

@ -410,21 +410,20 @@ void HeartbeatThread::runDBServer() {
/// CAS a key in the agency, works only if it does not exist, result should
/// contain the value of the written key.
static AgencyCommResult CasWithResult(AgencyComm agency, std::string const& key,
VPackSlice const& oldValue,
VPackSlice const& newJson, double timeout) {
static AgencyCommResult CasWithResult(
AgencyComm agency, std::string const& key, VPackSlice const& oldValue,
VPackSlice const& newJson, double timeout) {
AgencyOperation write(key, AgencyValueOperationType::SET, newJson);
write._ttl = 0; // no ttl
if (oldValue.isNone()) { // for some reason this doesn't work
// precondition: the key must equal old value
AgencyPrecondition pre(key, AgencyPrecondition::Type::EMPTY, true);
AgencyWriteTransaction trx(write, pre);
AgencyGeneralTransaction trx(write, pre);
return agency.sendTransactionWithFailover(trx, timeout);
} else {
// precondition: the key must equal old value
AgencyPrecondition pre(key, AgencyPrecondition::Type::VALUE, oldValue);
AgencyWriteTransaction trx(write, pre);
AgencyGeneralTransaction trx(write, pre);
return agency.sendTransactionWithFailover(trx, timeout);
}
}

37
js/client/tests/agency/agency-test.js
View File

@ -428,28 +428,29 @@ function agencyTestSuite () {
var pre = [{},{"a":12},{"a":12}];
cur += 2;
var wres = writeAndCheck([[query[0], pre[0], id[0]]]);
res = accessAgency("inquire",[id[0]]).bodyParsed;
assertEqual(res, {"results":[cur]});
assertEqual(res, wres);
wres = writeAndCheck([[query[1], pre[1], id[0]]]);
res = accessAgency("inquire",[id[0]]).bodyParsed;
assertEqual(res, {"results":[++cur]});
assertEqual(res, wres);
var wres = accessAgency("write", [[query[0], pre[0], id[0]]]);
res = accessAgency("inquire",[id[0]]);
wres.bodyParsed.inquired = true;
assertEqual(res.bodyParsed.results, wres.bodyParsed.results);
wres = accessAgency("write", [[query[1], pre[1], id[0]]]);
res = accessAgency("inquire",[id[0]]);
assertEqual(res.bodyParsed.results, wres.bodyParsed.results);
cur++;
wres = accessAgency("write",[[query[1], pre[1], id[2]]]);
assertEqual(wres.statusCode,412);
res = accessAgency("inquire",[id[2]]);
assertEqual(res.bodyParsed, {"results":[0]});
assertEqual(res, wres);
assertEqual(res.statusCode,412);
assertEqual(res.bodyParsed, {"results":[0],"inquired":true});
assertEqual(res.bodyParsed.results, wres.bodyParsed.results);
wres = accessAgency("write",[[query[0], pre[0], id[3]],
[query[1], pre[1], id[3]]]);
assertEqual(wres.statusCode,200);
cur += 2;
res = accessAgency("inquire",[id[3]]);
assertEqual(res.bodyParsed, {"results":[cur]});
assertEqual(res.bodyParsed, {"results":[cur],"inquired":true});
assertEqual(res.bodyParsed.results[0], wres.bodyParsed.results[1]);
assertEqual(res.statusCode,200);
@ -460,7 +461,7 @@ function agencyTestSuite () {
assertEqual(wres.statusCode,412);
cur += 2;
res = accessAgency("inquire",[id[4]]);
assertEqual(res.bodyParsed, {"results":[cur]});
assertEqual(res.bodyParsed, {"results":[cur],"inquired":true});
assertEqual(res.bodyParsed.results[0], wres.bodyParsed.results[1]);
assertEqual(res.statusCode,200);
@ -470,7 +471,7 @@ function agencyTestSuite () {
assertEqual(wres.statusCode,412);
cur += 2;
res = accessAgency("inquire",[id[5]]);
assertEqual(res.bodyParsed, {"results":[cur]});
assertEqual(res.bodyParsed, {"results":[cur],"inquired":true});
assertEqual(res.bodyParsed.results[0], wres.bodyParsed.results[1]);
assertEqual(res.statusCode,200);
@ -480,21 +481,21 @@ function agencyTestSuite () {
assertEqual(wres.statusCode,412);
cur += 2;
res = accessAgency("inquire",[id[6]]);
assertEqual(res.bodyParsed, {"results":[cur]});
assertEqual(res.bodyParsed, {"results":[cur],"inquired":true});
assertEqual(res.bodyParsed.results[0], wres.bodyParsed.results[2]);
assertEqual(res.statusCode,200);
wres = accessAgency("write",[[query[2], pre[2], id[7]],
[query[0], pre[0], id[8]],
[query[1], pre[1], id[9]]]);
assertEqual(res.statusCode,200);
assertEqual(wres.statusCode,412);
cur += 2;
res = accessAgency("inquire",[id[7],id[8],id[9]]);
assertEqual(res, wres);
assertEqual(res.statusCode,412);
assertEqual(res.bodyParsed.results, wres.bodyParsed.results);
},
////////////////////////////////////////////////////////////////////////////////
/// @brief test document/transaction assignment
////////////////////////////////////////////////////////////////////////////////