diff --git a/services/backup/src/Reactors/server/AddAttachmentsUtility.cpp b/services/backup/src/Reactors/server/AddAttachmentsUtility.cpp index 70faf7171..2e3c7586e 100644 --- a/services/backup/src/Reactors/server/AddAttachmentsUtility.cpp +++ b/services/backup/src/Reactors/server/AddAttachmentsUtility.cpp @@ -1,95 +1,97 @@ #include "AddAttachmentsUtility.h" +#include + #include "BackupItem.h" #include "BlobPutClientReactor.h" #include "Constants.h" #include "DatabaseManager.h" #include "ServiceBlobClient.h" #include "Tools.h" namespace comm { namespace network { namespace reactor { grpc::Status AddAttachmentsUtility::processRequest( const backup::AddAttachmentsRequest *request) { grpc::Status status = grpc::Status::OK; std::string userID = request->userid(); std::string backupID = request->backupid(); std::string logID = request->logid(); const std::string holders = request->holders(); try { if (userID.empty()) { throw std::runtime_error("user id required but not provided"); } if (backupID.empty()) { throw std::runtime_error("backup id required but not provided"); } if (holders.empty()) { throw std::runtime_error("holders required but not provided"); } if (logID.empty()) { // add these attachments to backup std::shared_ptr backupItem = database::DatabaseManager::getInstance().findBackupItem( userID, backupID); backupItem->addAttachmentHolders(holders); database::DatabaseManager::getInstance().putBackupItem(*backupItem); } else { // add these attachments to log std::shared_ptr logItem = database::DatabaseManager::getInstance().findLogItem(backupID, logID); logItem->addAttachmentHolders(holders); if (!logItem->getPersistedInBlob() && database::LogItem::getItemSize(logItem.get()) > LOG_DATA_SIZE_DATABASE_LIMIT) { bool old = logItem->getPersistedInBlob(); logItem = this->moveToS3(logItem); } database::DatabaseManager::getInstance().putLogItem(*logItem); } } catch (std::runtime_error &e) { - std::cout << "error: " << e.what() << std::endl; + LOG(ERROR) << e.what(); status = grpc::Status(grpc::StatusCode::INTERNAL, e.what()); } return status; } std::shared_ptr AddAttachmentsUtility::moveToS3(std::shared_ptr logItem) { std::string holder = tools::generateHolder( logItem->getDataHash(), logItem->getBackupID(), logItem->getLogID()); std::string data = std::move(logItem->getValue()); std::shared_ptr newLogItem = std::make_shared( logItem->getBackupID(), logItem->getLogID(), true, holder, logItem->getAttachmentHolders(), logItem->getDataHash()); // put into S3 std::condition_variable blobPutDoneCV; std::mutex blobPutDoneCVMutex; std::shared_ptr putReactor = std::make_shared( holder, newLogItem->getDataHash(), &blobPutDoneCV); ServiceBlobClient().put(putReactor); std::unique_lock lockPut(blobPutDoneCVMutex); putReactor->scheduleSendingDataChunk( std::make_unique(std::move(data))); putReactor->scheduleSendingDataChunk(std::make_unique("")); if (putReactor->getStatusHolder()->state != reactor::ReactorState::DONE) { blobPutDoneCV.wait(lockPut); } if (!putReactor->getStatusHolder()->getStatus().ok()) { throw std::runtime_error( putReactor->getStatusHolder()->getStatus().error_message()); } return newLogItem; } } // namespace reactor } // namespace network } // namespace comm diff --git a/services/backup/src/Reactors/server/CreateNewBackupReactor.cpp b/services/backup/src/Reactors/server/CreateNewBackupReactor.cpp index 04d08152f..73d29c219 100644 --- a/services/backup/src/Reactors/server/CreateNewBackupReactor.cpp +++ b/services/backup/src/Reactors/server/CreateNewBackupReactor.cpp @@ -1,113 +1,112 @@ #include "CreateNewBackupReactor.h" #include "DatabaseManager.h" #include "GlobalTools.h" #include "Tools.h" namespace comm { namespace network { namespace reactor { std::string CreateNewBackupReactor::generateBackupID() { if (this->deviceID.empty()) { throw std::runtime_error( "trying to generate a backup ID with an empty device ID"); } return this->deviceID + std::to_string(tools::getCurrentTimestamp()); } std::unique_ptr CreateNewBackupReactor::handleRequest( backup::CreateNewBackupRequest request, backup::CreateNewBackupResponse *response) { // we make sure that the blob client's state is flushed to the main memory // as there may be multiple threads from the pool taking over here const std::lock_guard lock(this->reactorStateMutex); switch (this->state) { case State::USER_ID: { if (!request.has_userid()) { throw std::runtime_error("user id expected but not received"); } this->userID = request.userid(); this->state = State::DEVICE_ID; return nullptr; } case State::DEVICE_ID: { if (!request.has_deviceid()) { throw std::runtime_error("device id expected but not received"); } this->deviceID = request.deviceid(); this->state = State::KEY_ENTROPY; return nullptr; } case State::KEY_ENTROPY: { if (!request.has_keyentropy()) { throw std::runtime_error( "backup key entropy expected but not received"); } this->keyEntropy = request.keyentropy(); this->state = State::DATA_HASH; return nullptr; } case State::DATA_HASH: { if (!request.has_newcompactionhash()) { throw std::runtime_error("data hash expected but not received"); } this->dataHash = request.newcompactionhash(); this->state = State::DATA_CHUNKS; this->backupID = this->generateBackupID(); if (database::DatabaseManager::getInstance().findBackupItem( this->userID, this->backupID) != nullptr) { throw std::runtime_error( "Backup with id [" + this->backupID + "] for user [" + this->userID + "] already exists, creation aborted"); } response->set_backupid(this->backupID); this->holder = tools::generateHolder(this->dataHash, this->backupID); this->putReactor = std::make_shared( this->holder, this->dataHash, &this->blobPutDoneCV); this->blobClient.put(this->putReactor); return nullptr; } case State::DATA_CHUNKS: { this->putReactor->scheduleSendingDataChunk(std::make_unique( std::move(*request.mutable_newcompactionchunk()))); return nullptr; } } throw std::runtime_error("new backup - invalid state"); } void CreateNewBackupReactor::terminateCallback() { const std::lock_guard lock(this->reactorStateMutex); if (this->putReactor == nullptr) { return; } this->putReactor->scheduleSendingDataChunk(std::make_unique("")); std::unique_lock lock2(this->blobPutDoneCVMutex); if (this->putReactor->getStatusHolder()->state != ReactorState::DONE) { this->blobPutDoneCV.wait(lock2); } if (this->putReactor->getStatusHolder()->state != ReactorState::DONE) { throw std::runtime_error("put reactor has not been terminated properly"); } if (!this->putReactor->getStatusHolder()->getStatus().ok()) { throw std::runtime_error( this->putReactor->getStatusHolder()->getStatus().error_message()); } // TODO add recovery data // TODO handle attachments holders database::BackupItem backupItem( this->userID, this->backupID, tools::getCurrentTimestamp(), tools::generateRandomString(), this->holder, {}); database::DatabaseManager::getInstance().putBackupItem(backupItem); - std::cout << "done creating backup " << backupItem.getBackupID() << std::endl; } } // namespace reactor } // namespace network } // namespace comm diff --git a/services/backup/src/Reactors/server/SendLogReactor.cpp b/services/backup/src/Reactors/server/SendLogReactor.cpp index b695fcaa1..c8fc8883e 100644 --- a/services/backup/src/Reactors/server/SendLogReactor.cpp +++ b/services/backup/src/Reactors/server/SendLogReactor.cpp @@ -1,179 +1,173 @@ #include "SendLogReactor.h" #include "Constants.h" #include "DatabaseManager.h" #include "GlobalTools.h" #include "Tools.h" -#include - namespace comm { namespace network { namespace reactor { void SendLogReactor::storeInDatabase() { bool storedInBlob = this->persistenceMethod == PersistenceMethod::BLOB; database::LogItem logItem( this->backupID, this->logID, storedInBlob, storedInBlob ? this->blobHolder : this->value, {}, this->hash); if (database::LogItem::getItemSize(&logItem) > LOG_DATA_SIZE_DATABASE_LIMIT) { throw std::runtime_error( "trying to put into the database an item with size " + std::to_string(database::LogItem::getItemSize(&logItem)) + " that exceeds the limit " + std::to_string(LOG_DATA_SIZE_DATABASE_LIMIT)); } database::DatabaseManager::getInstance().putLogItem(logItem); } std::string SendLogReactor::generateLogID(const std::string &backupID) { return backupID + tools::ID_SEPARATOR + std::to_string(tools::getCurrentTimestamp()); } void SendLogReactor::initializePutReactor() { if (this->blobHolder.empty()) { throw std::runtime_error( "put reactor cannot be initialized with empty blob holder"); } if (this->hash.empty()) { throw std::runtime_error( "put reactor cannot be initialized with empty hash"); } if (this->putReactor == nullptr) { this->putReactor = std::make_shared( this->blobHolder, this->hash, &this->blobPutDoneCV); this->blobClient.put(this->putReactor); } } std::unique_ptr SendLogReactor::readRequest(backup::SendLogRequest request) { // we make sure that the blob client's state is flushed to the main memory // as there may be multiple threads from the pool taking over here const std::lock_guard lock(this->reactorStateMutex); switch (this->state) { case State::USER_ID: { if (!request.has_userid()) { throw std::runtime_error("user id expected but not received"); } this->userID = request.userid(); this->state = State::BACKUP_ID; return nullptr; }; case State::BACKUP_ID: { if (!request.has_backupid()) { throw std::runtime_error("backup id expected but not received"); } this->backupID = request.backupid(); if (database::DatabaseManager::getInstance().findBackupItem( this->userID, this->backupID) == nullptr) { throw std::runtime_error( "trying to send log for a non-existent backup"); } this->logID = this->generateLogID(this->backupID); this->response->set_logcheckpoint(this->logID); this->state = State::LOG_HASH; return nullptr; }; case State::LOG_HASH: { if (!request.has_loghash()) { throw std::runtime_error("log hash expected but not received"); } this->hash = request.loghash(); this->state = State::LOG_CHUNK; return nullptr; }; case State::LOG_CHUNK: { if (!request.has_logdata()) { throw std::runtime_error("log data expected but not received"); } std::unique_ptr chunk = std::make_unique(std::move(*request.mutable_logdata())); if (chunk->size() == 0) { return std::make_unique(grpc::Status::OK); } if (this->persistenceMethod == PersistenceMethod::DB) { throw std::runtime_error( "please do not send multiple tiny chunks (less than " + std::to_string(LOG_DATA_SIZE_DATABASE_LIMIT) + "), merge them into bigger parts instead"); } if (this->persistenceMethod == PersistenceMethod::BLOB) { if (this->putReactor == nullptr) { throw std::runtime_error( "put reactor is being used but has not been initialized"); } this->putReactor->scheduleSendingDataChunk(std::move(chunk)); return nullptr; } this->value += std::move(*chunk); database::LogItem logItem = database::LogItem( this->backupID, this->logID, true, this->value, "", this->hash); if (database::LogItem::getItemSize(&logItem) > LOG_DATA_SIZE_DATABASE_LIMIT) { this->persistenceMethod = PersistenceMethod::BLOB; this->blobHolder = tools::generateHolder(this->hash, this->backupID, this->logID); this->initializePutReactor(); this->putReactor->scheduleSendingDataChunk( std::make_unique(this->value)); this->value = ""; } else { this->persistenceMethod = PersistenceMethod::DB; } return nullptr; }; } throw std::runtime_error("send log - invalid state"); } void SendLogReactor::terminateCallback() { const std::lock_guard lock(this->reactorStateMutex); if (!this->getStatusHolder()->getStatus().ok()) { throw std::runtime_error( this->getStatusHolder()->getStatus().error_message()); } if (this->persistenceMethod != PersistenceMethod::BLOB && this->persistenceMethod != PersistenceMethod::DB) { throw std::runtime_error("Invalid persistence method detected"); } if (this->persistenceMethod == PersistenceMethod::DB || this->putReactor == nullptr) { this->storeInDatabase(); return; } this->putReactor->scheduleSendingDataChunk(std::make_unique("")); std::unique_lock lockPut(this->blobPutDoneCVMutex); if (this->putReactor->getStatusHolder()->state != ReactorState::DONE) { this->blobPutDoneCV.wait(lockPut); } if (!this->putReactor->getStatusHolder()->getStatus().ok()) { throw std::runtime_error( this->putReactor->getStatusHolder()->getStatus().error_message()); } // store in db only when we successfully upload chunks this->storeInDatabase(); } void SendLogReactor::doneCallback() { // we make sure that the blob client's state is flushed to the main memory // as there may be multiple threads from the pool taking over here const std::lock_guard lock(this->reactorStateMutex); // TODO implement - std::cout << "receive logs done " - << this->getStatusHolder()->getStatus().error_code() << "/" - << this->getStatusHolder()->getStatus().error_message() - << std::endl; } } // namespace reactor } // namespace network } // namespace comm diff --git a/services/backup/src/server.cpp b/services/backup/src/server.cpp index 67b37868f..884da8559 100644 --- a/services/backup/src/server.cpp +++ b/services/backup/src/server.cpp @@ -1,39 +1,42 @@ #include "BackupServiceImpl.h" +#include "GlobalTools.h" + +#include #include -#include #include #include namespace comm { namespace network { void RunServer() { std::string server_address = "0.0.0.0:50051"; BackupServiceImpl backupService; grpc::EnableDefaultHealthCheckService(true); grpc::ServerBuilder builder; // Listen on the given address without any authentication mechanism. builder.AddListeningPort(server_address, grpc::InsecureServerCredentials()); // Register "service" as the instance through which we'll communicate with // clients. In this case it corresponds to an *synchronous* service. builder.RegisterService(&backupService); // Finally assemble the server. std::unique_ptr server(builder.BuildAndStart()); - std::cout << "Server listening" << std::endl; + LOG(INFO) << "Server listening"; // Wait for the server to shutdown. Note that some other thread must be // responsible for shutting down the server for this call to ever return. server->Wait(); } } // namespace network } // namespace comm int main(int argc, char **argv) { + comm::network::tools::InitLogging("backup"); comm::network::RunServer(); return 0; }