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

Merge branch 'devel' of https://github.com/arangodb/arangodb into generic-col-types

This commit is contained in:
jsteemann 2016-09-07 12:06:43 +02:00
parent 96eabfc64a 288f42c531
commit dfca849325
3 changed files with 78 additions and 68 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

3
arangod/Agency/Agent.cpp
View File

@ -136,7 +136,6 @@ void Agent::startConstituent() {
auto database = ApplicationServer::getFeature<DatabaseFeature>("Database"); auto database = ApplicationServer::getFeature<DatabaseFeature>("Database");
auto vocbase = database->systemDatabase(); auto vocbase = database->systemDatabase();
auto queryRegistry = QueryRegistryFeature::QUERY_REGISTRY; auto queryRegistry = QueryRegistryFeature::QUERY_REGISTRY;
_constituent.start(vocbase, queryRegistry);
} }
// Waits here for confirmation of log's commits up to index. // Waits here for confirmation of log's commits up to index.
@ -402,8 +401,8 @@ bool Agent::load() {
TRI_ASSERT(queryRegistry != nullptr); TRI_ASSERT(queryRegistry != nullptr);
if (size() == 1) { if (size() == 1) {
activateAgency(); activateAgency();
_constituent.start(vocbase, queryRegistry);
} }
_constituent.start(vocbase, queryRegistry);
if (_config.supervision()) { if (_config.supervision()) {
LOG_TOPIC(DEBUG, Logger::AGENCY) << "Starting cluster sanity facilities"; LOG_TOPIC(DEBUG, Logger::AGENCY) << "Starting cluster sanity facilities";

2
arangod/Agency/Constituent.cpp
View File

@ -229,7 +229,7 @@ std::string Constituent::endpoint(std::string id) const {
/// @brief Vote /// @brief Vote
bool Constituent::vote(term_t term, std::string id, index_t prevLogIndex, bool Constituent::vote(term_t term, std::string id, index_t prevLogIndex,
term_t prevLogTerm, bool appendEntries) { term_t prevLogTerm, bool appendEntries) {
if(!_vocbase) { if(_vocbase==nullptr) {
return false; return false;
} }

141
arangod/Cluster/ClusterComm.cpp
View File

@ -68,16 +68,16 @@ void ClusterCommResult::setDestination(std::string const& dest,
serverID = ""; serverID = "";
status = CL_COMM_BACKEND_UNAVAILABLE; status = CL_COMM_BACKEND_UNAVAILABLE;
if (logConnectionErrors) { if (logConnectionErrors) {
LOG(ERR) << "cannot find responsible server for shard '" << shardID LOG_TOPIC(ERR, Logger::CLUSTER)
<< "'"; << "cannot find responsible server for shard '" << shardID << "'";
} else { } else {
LOG(INFO) << "cannot find responsible server for shard '" << shardID LOG_TOPIC(INFO, Logger::CLUSTER)
<< "'"; << "cannot find responsible server for shard '" << shardID << "'";
} }
return; return;
} }
} }
LOG(DEBUG) << "Responsible server: " << serverID; LOG_TOPIC(DEBUG, Logger::CLUSTER) << "Responsible server: " << serverID;
} else if (dest.substr(0, 7) == "server:") { } else if (dest.substr(0, 7) == "server:") {
shardID = ""; shardID = "";
serverID = dest.substr(7); serverID = dest.substr(7);
@ -93,9 +93,11 @@ void ClusterCommResult::setDestination(std::string const& dest,
status = CL_COMM_BACKEND_UNAVAILABLE; status = CL_COMM_BACKEND_UNAVAILABLE;
errorMessage = "did not understand destination'" + dest + "'"; errorMessage = "did not understand destination'" + dest + "'";
if (logConnectionErrors) { if (logConnectionErrors) {
LOG(ERR) << "did not understand destination '" << dest << "'"; LOG_TOPIC(ERR, Logger::CLUSTER)
<< "did not understand destination '" << dest << "'";
} else { } else {
LOG(INFO) << "did not understand destination '" << dest << "'"; LOG_TOPIC(INFO, Logger::CLUSTER)
<< "did not understand destination '" << dest << "'";
} }
return; return;
} }
@ -106,9 +108,11 @@ void ClusterCommResult::setDestination(std::string const& dest,
status = CL_COMM_BACKEND_UNAVAILABLE; status = CL_COMM_BACKEND_UNAVAILABLE;
errorMessage = "did not find endpoint of server '" + serverID + "'"; errorMessage = "did not find endpoint of server '" + serverID + "'";
if (logConnectionErrors) { if (logConnectionErrors) {
LOG(ERR) << "did not find endpoint of server '" << serverID << "'"; LOG_TOPIC(ERR, Logger::CLUSTER)
<< "did not find endpoint of server '" << serverID << "'";
} else { } else {
LOG(INFO) << "did not find endpoint of server '" << serverID << "'"; LOG_TOPIC(INFO, Logger::CLUSTER)
<< "did not find endpoint of server '" << serverID << "'";
} }
} }
} }
@ -251,7 +255,8 @@ void ClusterComm::startBackgroundThread() {
_backgroundThread = new ClusterCommThread(); _backgroundThread = new ClusterCommThread();
if (!_backgroundThread->start()) { if (!_backgroundThread->start()) {
LOG(FATAL) << "ClusterComm background thread does not work"; LOG_TOPIC(FATAL, Logger::CLUSTER)
<< "ClusterComm background thread does not work";
FATAL_ERROR_EXIT(); FATAL_ERROR_EXIT();
} }
} }
@ -348,8 +353,9 @@ OperationID ClusterComm::asyncRequest(
if (op->result.status == CL_COMM_BACKEND_UNAVAILABLE) { if (op->result.status == CL_COMM_BACKEND_UNAVAILABLE) {
// We put it into the received queue right away for error reporting: // We put it into the received queue right away for error reporting:
ClusterCommResult const resCopy(op->result); ClusterCommResult const resCopy(op->result);
LOG(DEBUG) << "In asyncRequest, putting failed request " LOG_TOPIC(DEBUG, Logger::CLUSTER)
<< resCopy.operationID << " directly into received queue."; << "In asyncRequest, putting failed request " << resCopy.operationID
<< " directly into received queue.";
CONDITION_LOCKER(locker, somethingReceived); CONDITION_LOCKER(locker, somethingReceived);
received.push_back(op.get()); received.push_back(op.get());
op.release(); op.release();
@ -411,8 +417,8 @@ OperationID ClusterComm::asyncRequest(
// LOCKING-DEBUG // LOCKING-DEBUG
// std::cout << "asyncRequest: sending " << // std::cout << "asyncRequest: sending " <<
// arangodb::rest::HttpRequest::translateMethod(reqtype) << " request to DB // arangodb::rest::HttpRequest::translateMethod(reqtype, Logger::CLUSTER) << " request to DB
// server '" << op->serverID << ":" << path << "\n" << *(body.get()) << "\n"; // server '" << op->serverID << ":" << path << "\n" << *(body.get(), Logger::CLUSTER) << "\n";
// for (auto& h : *(op->headerFields)) { // for (auto& h : *(op->headerFields)) {
// std::cout << h.first << ":" << h.second << std::endl; // std::cout << h.first << ":" << h.second << std::endl;
// } // }
@ -426,7 +432,8 @@ OperationID ClusterComm::asyncRequest(
std::list<ClusterCommOperation*>::iterator i = toSend.end(); std::list<ClusterCommOperation*>::iterator i = toSend.end();
toSendByOpID[opId] = --i; toSendByOpID[opId] = --i;
} }
LOG(DEBUG) << "In asyncRequest, put into queue " << opId; LOG_TOPIC(DEBUG, Logger::CLUSTER)
<< "In asyncRequest, put into queue " << opId;
somethingToSend.signal(); somethingToSend.signal();
return opId; return opId;
@ -501,19 +508,22 @@ std::unique_ptr<ClusterCommResult> ClusterComm::syncRequest(
res->errorMessage = res->errorMessage =
"cannot create connection to server '" + res->serverID + "'"; "cannot create connection to server '" + res->serverID + "'";
if (logConnectionErrors()) { if (logConnectionErrors()) {
LOG(ERR) << "cannot create connection to server '" << res->serverID LOG_TOPIC(ERR, Logger::CLUSTER)
<< "' at endpoint '" << res->endpoint << "'"; << "cannot create connection to server '" << res->serverID
<< "' at endpoint '" << res->endpoint << "'";
} else { } else {
LOG(INFO) << "cannot create connection to server '" << res->serverID LOG_TOPIC(INFO, Logger::CLUSTER)
<< "' at endpoint '" << res->endpoint << "'"; << "cannot create connection to server '" << res->serverID
<< "' at endpoint '" << res->endpoint << "'";
} }
} else { } else {
LOG(DEBUG) << "sending " << arangodb::HttpRequest::translateMethod(reqtype) LOG_TOPIC(DEBUG, Logger::CLUSTER)
<< " request to DB server '" << res->serverID << "sending " << arangodb::HttpRequest::translateMethod(reqtype)
<< "' at endpoint '" << res->endpoint << "': " << body; << " request to DB server '" << res->serverID << "' at endpoint '"
<< res->endpoint << "': " << body;
// LOCKING-DEBUG // LOCKING-DEBUG
// std::cout << "syncRequest: sending " << // std::cout << "syncRequest: sending " <<
// arangodb::rest::HttpRequest::translateMethod(reqtype) << " request to // arangodb::rest::HttpRequest::translateMethod(reqtype, Logger::CLUSTER) << " request to
// DB server '" << res->serverID << ":" << path << "\n" << body << "\n"; // DB server '" << res->serverID << ":" << path << "\n" << body << "\n";
// for (auto& h : headersCopy) { // for (auto& h : headersCopy) {
// std::cout << h.first << ":" << h.second << std::endl; // std::cout << h.first << ":" << h.second << std::endl;
@ -888,11 +898,13 @@ void ClusterComm::asyncAnswer(std::string& coordinatorHeader,
size_t start = 0; size_t start = 0;
size_t pos; size_t pos;
LOG(DEBUG) << "In asyncAnswer, seeing " << coordinatorHeader; LOG_TOPIC(DEBUG, Logger::CLUSTER)
<< "In asyncAnswer, seeing " << coordinatorHeader;
pos = coordinatorHeader.find(":", start); pos = coordinatorHeader.find(":", start);
if (pos == std::string::npos) { if (pos == std::string::npos) {
LOG(ERR) << "Could not find coordinator ID in X-Arango-Coordinator"; LOG_TOPIC(ERR, Logger::CLUSTER)
<< "Could not find coordinator ID in X-Arango-Coordinator";
return; return;
} }
@ -905,11 +917,13 @@ void ClusterComm::asyncAnswer(std::string& coordinatorHeader,
if (endpoint == "") { if (endpoint == "") {
if (logConnectionErrors()) { if (logConnectionErrors()) {
LOG(ERR) << "asyncAnswer: cannot find endpoint for server '" LOG_TOPIC(ERR, Logger::CLUSTER)
<< coordinatorID << "'"; << "asyncAnswer: cannot find endpoint for server '"
<< coordinatorID << "'";
} else { } else {
LOG(INFO) << "asyncAnswer: cannot find endpoint for server '" LOG_TOPIC(INFO, Logger::CLUSTER)
<< coordinatorID << "'"; << "asyncAnswer: cannot find endpoint for server '"
<< coordinatorID << "'";
} }
return; return;
} }
@ -918,8 +932,9 @@ void ClusterComm::asyncAnswer(std::string& coordinatorHeader,
cm->leaseConnection(endpoint); cm->leaseConnection(endpoint);
if (nullptr == connection) { if (nullptr == connection) {
LOG(ERR) << "asyncAnswer: cannot create connection to server '" LOG_TOPIC(ERR, Logger::CLUSTER)
<< coordinatorID << "'"; << "asyncAnswer: cannot create connection to server '"
<< coordinatorID << "'";
return; return;
} }
@ -936,8 +951,9 @@ void ClusterComm::asyncAnswer(std::string& coordinatorHeader,
char const* body = responseToSend->body().c_str(); char const* body = responseToSend->body().c_str();
size_t len = responseToSend->body().length(); size_t len = responseToSend->body().length();
LOG(DEBUG) << "asyncAnswer: sending PUT request to DB server '" LOG_TOPIC(DEBUG, Logger::CLUSTER)
<< coordinatorID << "'"; << "asyncAnswer: sending PUT request to DB server '"
<< coordinatorID << "'";
auto client = std::make_unique<arangodb::httpclient::SimpleHttpClient>( auto client = std::make_unique<arangodb::httpclient::SimpleHttpClient>(
connection->_connection, 3600.0, false); connection->_connection, 3600.0, false);
@ -978,7 +994,8 @@ std::string ClusterComm::processAnswer(
size_t start = 0; size_t start = 0;
size_t pos; size_t pos;
LOG(DEBUG) << "In processAnswer, seeing " << coordinatorHeader; LOG_TOPIC(DEBUG, Logger::CLUSTER)
<< "In processAnswer, seeing " << coordinatorHeader;
pos = coordinatorHeader.find(":", start); pos = coordinatorHeader.find(":", start);
if (pos == std::string::npos) { if (pos == std::string::npos) {
@ -1058,7 +1075,7 @@ std::string ClusterComm::processAnswer(
//////////////////////////////////////////////////////////////////////////////// ////////////////////////////////////////////////////////////////////////////////
bool ClusterComm::moveFromSendToReceived(OperationID operationID) { bool ClusterComm::moveFromSendToReceived(OperationID operationID) {
LOG(DEBUG) << "In moveFromSendToReceived " << operationID; LOG_TOPIC(DEBUG, Logger::CLUSTER) << "In moveFromSendToReceived " << operationID;
CONDITION_LOCKER(locker, somethingReceived); CONDITION_LOCKER(locker, somethingReceived);
CONDITION_LOCKER(sendLocker, somethingToSend); CONDITION_LOCKER(sendLocker, somethingToSend);
@ -1250,8 +1267,7 @@ size_t ClusterComm::performRequests(std::vector<ClusterCommRequest>& requests,
auto it = opIDtoIndex.find(res.operationID); auto it = opIDtoIndex.find(res.operationID);
if (it == opIDtoIndex.end()) { if (it == opIDtoIndex.end()) {
// Ooops, we got a response to which we did not send the request // Ooops, we got a response to which we did not send the request
LOG(ERR) LOG_TOPIC(ERR, Logger::CLUSTER) << "Received ClusterComm response for a request we did not send!";
<< "Received ClusterComm response for a request we did not send!";
continue; continue;
} }
size_t index = it->second; size_t index = it->second;
@ -1264,8 +1280,7 @@ size_t ClusterComm::performRequests(std::vector<ClusterCommRequest>& requests,
res.answer_code == rest::ResponseCode::ACCEPTED) { res.answer_code == rest::ResponseCode::ACCEPTED) {
nrGood++; nrGood++;
} }
LOG_TOPIC(TRACE, logTopic) LOG_TOPIC(TRACE, Logger::CLUSTER) << "ClusterComm::performRequests: "
<< "ClusterComm::performRequests: "
<< "got answer from " << requests[index].destination << ":" << "got answer from " << requests[index].destination << ":"
<< requests[index].path << " with return code " << requests[index].path << " with return code "
<< (int)res.answer_code; << (int)res.answer_code;
@ -1283,16 +1298,14 @@ size_t ClusterComm::performRequests(std::vector<ClusterCommRequest>& requests,
if (dueTime[index] < actionNeeded) { if (dueTime[index] < actionNeeded) {
actionNeeded = dueTime[index]; actionNeeded = dueTime[index];
} }
LOG_TOPIC(TRACE, logTopic) LOG_TOPIC(TRACE, Logger::CLUSTER) << "ClusterComm::performRequests: "
<< "ClusterComm::performRequests: "
<< "got BACKEND_UNAVAILABLE or TIMEOUT from " << "got BACKEND_UNAVAILABLE or TIMEOUT from "
<< requests[index].destination << ":" << requests[index].path; << requests[index].destination << ":" << requests[index].path;
} else { // a "proper error" } else { // a "proper error"
requests[index].result = res; requests[index].result = res;
requests[index].done = true; requests[index].done = true;
nrDone++; nrDone++;
LOG_TOPIC(TRACE, logTopic) LOG_TOPIC(TRACE, Logger::CLUSTER) << "ClusterComm::performRequests: "
<< "ClusterComm::performRequests: "
<< "got no answer from " << requests[index].destination << ":" << "got no answer from " << requests[index].destination << ":"
<< requests[index].path << " with error " << res.status; << requests[index].path << " with error " << res.status;
} }
@ -1303,8 +1316,8 @@ size_t ClusterComm::performRequests(std::vector<ClusterCommRequest>& requests,
} }
} }
} catch (...) { } catch (...) {
LOG_TOPIC(ERR, logTopic) << "ClusterComm::performRequests: " LOG_TOPIC(ERR, Logger::CLUSTER) << "ClusterComm::performRequests: "
<< "caught exception, ignoring..."; << "caught exception, ignoring...";
} }
// We only get here if the global timeout was triggered, not all // We only get here if the global timeout was triggered, not all
@ -1405,7 +1418,7 @@ void ClusterCommThread::run() {
ClusterCommOperation* op; ClusterCommOperation* op;
ClusterComm* cc = ClusterComm::instance(); ClusterComm* cc = ClusterComm::instance();
LOG(DEBUG) << "starting ClusterComm thread"; LOG_TOPIC(DEBUG, Logger::CLUSTER) << "starting ClusterComm thread";
while (!isStopping()) { while (!isStopping()) {
// First check the sending queue, as long as it is not empty, we send // First check the sending queue, as long as it is not empty, we send
@ -1421,7 +1434,7 @@ void ClusterCommThread::run() {
if (cc->toSend.empty()) { if (cc->toSend.empty()) {
break; break;
} else { } else {
LOG(DEBUG) << "Noticed something to send"; LOG_TOPIC(DEBUG, Logger::CLUSTER) << "Noticed something to send";
op = cc->toSend.front(); op = cc->toSend.front();
TRI_ASSERT(op->result.status == CL_COMM_SUBMITTED); TRI_ASSERT(op->result.status == CL_COMM_SUBMITTED);
op->result.status = CL_COMM_SENDING; op->result.status = CL_COMM_SENDING;
@ -1448,28 +1461,26 @@ void ClusterCommThread::run() {
op->result.errorMessage = "cannot create connection to server: "; op->result.errorMessage = "cannot create connection to server: ";
op->result.errorMessage += op->result.serverID; op->result.errorMessage += op->result.serverID;
if (cc->logConnectionErrors()) { if (cc->logConnectionErrors()) {
LOG(ERR) << "cannot create connection to server '" LOG_TOPIC(ERR, Logger::CLUSTER) << "cannot create connection to server '"
<< op->result.serverID << "' at endpoint '" << op->result.serverID << "' at endpoint '" << op->result.endpoint << "'";
<< op->result.endpoint << "'";
} else { } else {
LOG(INFO) << "cannot create connection to server '" LOG_TOPIC(INFO, Logger::CLUSTER) << "cannot create connection to server '"
<< op->result.serverID << "' at endpoint '" << op->result.serverID << "' at endpoint '" << op->result.endpoint << "'";
<< op->result.endpoint << "'";
} }
} else { } else {
if (nullptr != op->body.get()) { if (nullptr != op->body.get()) {
LOG(DEBUG) LOG_TOPIC(DEBUG, Logger::CLUSTER) << "sending "
<< "sending " << arangodb::HttpRequest::translateMethod(
<< arangodb::HttpRequest::translateMethod(op->reqtype).c_str() op->reqtype)
<< " request to DB server '" << op->result.serverID << " request to DB server '"
<< "' at endpoint '" << op->result.endpoint << op->result.serverID << "' at endpoint '" << op->result.endpoint
<< "': " << op->body->c_str(); << "': " << std::string(op->body->c_str(), op->body->size());
} else { } else {
LOG(DEBUG) LOG_TOPIC(DEBUG, Logger::CLUSTER) << "sending "
<< "sending " << arangodb::HttpRequest::translateMethod(
<< arangodb::HttpRequest::translateMethod(op->reqtype).c_str() op->reqtype)
<< " request to DB server '" << op->result.serverID << " request to DB server '"
<< "' at endpoint '" << op->result.endpoint << "'"; << op->result.serverID << "' at endpoint '" << op->result.endpoint << "'";
} }
auto client = auto client =
@ -1586,5 +1597,5 @@ void ClusterCommThread::run() {
} }
} }
LOG(DEBUG) << "stopped ClusterComm thread"; LOG_TOPIC(DEBUG, Logger::CLUSTER) << "stopped ClusterComm thread";
} }