diff --git a/src/replica/apps/WorkerApp.cc b/src/replica/apps/WorkerApp.cc index 31c023640..f1362b91a 100644 --- a/src/replica/apps/WorkerApp.cc +++ b/src/replica/apps/WorkerApp.cc @@ -39,7 +39,7 @@ #include "replica/services/ServiceProvider.h" #include "replica/util/FileUtils.h" #include "replica/worker/FileServer.h" -#include "replica/worker/WorkerProcessor.h" +#include "replica/worker/WorkerHttpSvc.h" #include "replica/worker/WorkerServer.h" // LSST headers @@ -113,6 +113,9 @@ int WorkerApp::runImpl() { auto const reqProcSvr = WorkerServer::create(serviceProvider(), worker); thread reqProcSvrThread([reqProcSvr]() { reqProcSvr->run(); }); + auto const reqProcHttpSvr = WorkerHttpSvc::create(serviceProvider(), worker); + thread reqProcHttpSvrThread([reqProcHttpSvr]() { reqProcHttpSvr->run(); }); + auto const fileSvr = FileServer::create(serviceProvider(), worker); thread fileSvrThread([fileSvr]() { fileSvr->run(); }); diff --git a/src/replica/proto/CMakeLists.txt b/src/replica/proto/CMakeLists.txt index b61599d8c..7eb8d830d 100644 --- a/src/replica/proto/CMakeLists.txt +++ b/src/replica/proto/CMakeLists.txt @@ -4,4 +4,5 @@ add_library(replica_proto OBJECT) target_sources(replica_proto PRIVATE ${REPLICA_PB_SRCS} ${REPLICA_PB_HDRS} + Protocol.cc ) diff --git a/src/replica/proto/Protocol.cc b/src/replica/proto/Protocol.cc new file mode 100644 index 000000000..3c4b878c3 --- /dev/null +++ b/src/replica/proto/Protocol.cc @@ -0,0 +1,134 @@ +/* + * LSST Data Management System + * + * This product includes software developed by the + * LSST Project (http://www.lsst.org/). + * + * This program is free software: you can redistribute it and/or modify + * it under the terms of the GNU General Public License as published by + * the Free Software Foundation, either version 3 of the License, or + * (at your option) any later version. + * + * This program is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * GNU General Public License for more details. + * + * You should have received a copy of the LSST License Statement and + * the GNU General Public License along with this program. If not, + * see . + */ + +// Class header +#include "replica/proto/Protocol.h" + +// System headers +#include + +using namespace std; + +namespace lsst::qserv::replica::protocol { + +std::string toString(Status status) { + switch (status) { + case Status::CREATED: + return "CREATED"; + case Status::SUCCESS: + return "SUCCESS"; + case Status::QUEUED: + return "QUEUED"; + case Status::IN_PROGRESS: + return "IN_PROGRESS"; + case Status::IS_CANCELLING: + return "IS_CANCELLING"; + case Status::BAD: + return "BAD"; + case Status::FAILED: + return "FAILED"; + case Status::CANCELLED: + return "CANCELLED"; + default: + throw logic_error("Unhandled status: " + to_string(static_cast(status))); + } +} + +std::string toString(StatusExt extendedStatus) { + switch (extendedStatus) { + case StatusExt::NONE: + return "NONE"; + case StatusExt::INVALID_PARAM: + return "INVALID_PARAM"; + case StatusExt::INVALID_ID: + return "INVALID_ID"; + case StatusExt::FOLDER_STAT: + return "FOLDER_STAT"; + case StatusExt::FOLDER_CREATE: + return "FOLDER_CREATE"; + case StatusExt::FILE_STAT: + return "FILE_STAT"; + case StatusExt::FILE_SIZE: + return "FILE_SIZE"; + case StatusExt::FOLDER_READ: + return "FOLDER_READ"; + case StatusExt::FILE_READ: + return "FILE_READ"; + case StatusExt::FILE_ROPEN: + return "FILE_ROPEN"; + case StatusExt::FILE_CREATE: + return "FILE_CREATE"; + case StatusExt::FILE_OPEN: + return "FILE_OPEN"; + case StatusExt::FILE_RESIZE: + return "FILE_RESIZE"; + case StatusExt::FILE_WRITE: + return "FILE_WRITE"; + case StatusExt::FILE_COPY: + return "FILE_COPY"; + case StatusExt::FILE_DELETE: + return "FILE_DELETE"; + case StatusExt::FILE_RENAME: + return "FILE_RENAME"; + case StatusExt::FILE_EXISTS: + return "FILE_EXISTS"; + case StatusExt::SPACE_REQ: + return "SPACE_REQ"; + case StatusExt::NO_FOLDER: + return "NO_FOLDER"; + case StatusExt::NO_FILE: + return "NO_FILE"; + case StatusExt::NO_ACCESS: + return "NO_ACCESS"; + case StatusExt::NO_SPACE: + return "NO_SPACE"; + case StatusExt::FILE_MTIME: + return "FILE_MTIME"; + case StatusExt::MYSQL_ERROR: + return "MYSQL_ERROR"; + case StatusExt::LARGE_RESULT: + return "LARGE_RESULT"; + case StatusExt::NO_SUCH_TABLE: + return "NO_SUCH_TABLE"; + case StatusExt::NOT_PARTITIONED_TABLE: + return "NOT_PARTITIONED_TABLE"; + case StatusExt::NO_SUCH_PARTITION: + return "NO_SUCH_PARTITION"; + case StatusExt::MULTIPLE: + return "MULTIPLE"; + case StatusExt::OTHER_EXCEPTION: + return "OTHER_EXCEPTION"; + case StatusExt::FOREIGN_INSTANCE: + return "FOREIGN_INSTANCE"; + case StatusExt::DUPLICATE_KEY: + return "DUPLICATE_KEY"; + case StatusExt::CANT_DROP_KEY: + return "CANT_DROP_KEY"; + default: + throw logic_error("Unhandled extended status: " + to_string(static_cast(extendedStatus))); + } +} + +string toString(Status status, StatusExt extendedStatus) { + return toString(status) + "::" + toString(extendedStatus); +} + +} // namespace lsst::qserv::replica::protocol diff --git a/src/replica/proto/Protocol.h b/src/replica/proto/Protocol.h new file mode 100644 index 000000000..42ddf12e2 --- /dev/null +++ b/src/replica/proto/Protocol.h @@ -0,0 +1,136 @@ +/* + * LSST Data Management System + * + * This product includes software developed by the + * LSST Project (http://www.lsst.org/). + * + * This program is free software: you can redistribute it and/or modify + * it under the terms of the GNU General Public License as published by + * the Free Software Foundation, either version 3 of the License, or + * (at your option) any later version. + * + * This program is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * GNU General Public License for more details. + * + * You should have received a copy of the LSST License Statement and + * the GNU General Public License along with this program. If not, + * see . + */ +#ifndef LSST_QSERV_REPLICA_PROTOCOL_H +#define LSST_QSERV_REPLICA_PROTOCOL_H + +// System headers +#include + +// Third party headers +#include "nlohmann/json.hpp" + +// This header declarations +namespace lsst::qserv::replica::protocol { + +/// Subtypes of the SQL requests. +enum class SqlRequestType : int { + + QUERY = 0, + CREATE_DATABASE = 1, + DROP_DATABASE = 2, + ENABLE_DATABASE = 3, ///< in Qserv + DISABLE_DATABASE = 4, ///< in Qserv + GRANT_ACCESS = 5, + CREATE_TABLE = 6, + DROP_TABLE = 7, + REMOVE_TABLE_PARTITIONING = 8, + DROP_TABLE_PARTITION = 9, + GET_TABLE_INDEX = 10, + CREATE_TABLE_INDEX = 11, + DROP_TABLE_INDEX = 12, + ALTER_TABLE = 13, + TABLE_ROW_STATS = 14 +}; + +/// Types of the table indexes specified in the index management requests requests. +enum class SqlIndexSpec : int { DEFAULT = 1, UNIQUE = 2, FULLTEXT = 3, SPATIAL = 4 }; + +/// Status values returned by all request related to operations with +/// replicas. Request management operations always return messages whose types +/// match the return types of the corresponding (original) replica-related requests. +/// Service management requests have their own set of status values. +/// +enum class Status : int { + CREATED = 0, + SUCCESS = 1, + QUEUED = 2, + IN_PROGRESS = 3, + IS_CANCELLING = 4, + BAD = 5, + FAILED = 6, + CANCELLED = 7 +}; + +enum class StatusExt : int { + NONE = 0, ///< Unspecified problem. + INVALID_PARAM = 1, ///< Invalid parameter(s) of a request. + INVALID_ID = 2, ///< An invalid request identifier. + FOLDER_STAT = 4, ///< Failed to obtain fstat() for a folder. + FOLDER_CREATE = 5, ///< Failed to create a folder. + FILE_STAT = 6, ///< Failed to obtain fstat() for a file. + FILE_SIZE = 7, ///< Failed to obtain a size of a file. + FOLDER_READ = 8, ///< Failed to read the contents of a folder. + FILE_READ = 9, ///< Failed to read the contents of a file. + FILE_ROPEN = 10, ///< Failed to open a remote file. + FILE_CREATE = 11, ///< Failed to create a file. + FILE_OPEN = 12, ///< Failed to open a file. + FILE_RESIZE = 13, ///< Failed to resize a file. + FILE_WRITE = 14, ///< Failed to write into a file. + FILE_COPY = 15, ///< Failed to copy a file. + FILE_DELETE = 16, ///< Failed to delete a file. + FILE_RENAME = 17, ///< Failed to rename a file. + FILE_EXISTS = 18, ///< File already exists. + SPACE_REQ = 19, ///< Space availability check failed. + NO_FOLDER = 20, ///< Folder doesn't exist. + NO_FILE = 21, ///< File doesn't exist. + NO_ACCESS = 22, ///< No access to a file or a folder. + NO_SPACE = 23, ///< No space left on a device as required by an operation. + FILE_MTIME = 24, ///< Get/set 'mtime' operation failed. + MYSQL_ERROR = 25, ///< General MySQL error (other than any specific ones listed here). + LARGE_RESULT = 26, ///< Result exceeds a limit set in a request. + NO_SUCH_TABLE = 27, ///< No table found while performing a MySQL operation. + NOT_PARTITIONED_TABLE = 28, ///< The table is not MySQL partitioned as it was expected. + NO_SUCH_PARTITION = 29, ///< No MySQL partition found in a table as it was expected. + MULTIPLE = 30, ///< Multiple unspecified errors encountered when processing a request. + OTHER_EXCEPTION = 31, ///< Other exception not listed here. + FOREIGN_INSTANCE = 32, ///< Detected a request from a Controller serving an unrelated Qserv. + DUPLICATE_KEY = 33, ///< Duplicate key found when creating an index or altering a table schema. + CANT_DROP_KEY = 34 ///< Can't drop a field or a key which doesn't exist. +}; + +/// @return the string representation of the status +std::string toString(Status status); + +/// @return the string representation of the extended status +std::string toString(StatusExt extendedStatus); + +/// @return the string representation of the full status +std::string toString(Status status, StatusExt extendedStatus); + +/// Status of a replica. +enum class ReplicaStatus : int { NOT_FOUND = 0, CORRUPT = 1, INCOMPLETE = 2, COMPLETE = 3 }; + +/// Status of a service. +enum class ServiceState : int { SUSPEND_IN_PROGRESS = 0, SUSPENDED = 1, RUNNING = 2 }; + +/// The header to be sent with the requests processed through the worker's queueing system. +struct QueuedRequestHdr { + std::string id; + int priority; + unsigned int timeout; + QueuedRequestHdr(std::string const& id_, int priority_, unsigned int timeout_) + : id(id_), priority(priority_), timeout(timeout_) {} + nlohmann::json toJson() const { return {{"id", id}, {"priority", priority}, {"timeout", timeout}}; }; +}; + +} // namespace lsst::qserv::replica::protocol + +#endif // LSST_QSERV_REPLICA_PROTOCOL_H diff --git a/src/replica/util/Performance.cc b/src/replica/util/Performance.cc index 8e3292d68..ae30b0ac3 100644 --- a/src/replica/util/Performance.cc +++ b/src/replica/util/Performance.cc @@ -30,6 +30,7 @@ #include "lsst/log/Log.h" using namespace std; +using json = nlohmann::json; namespace { diff --git a/src/replica/util/Performance.h b/src/replica/util/Performance.h index fcbfd394a..15320d08b 100644 --- a/src/replica/util/Performance.h +++ b/src/replica/util/Performance.h @@ -33,6 +33,9 @@ #include #include +// Third party headers +#include "nlohmann/json.hpp" + // Forward declarations namespace lsst::qserv::replica { class ProtocolPerformance; @@ -56,7 +59,6 @@ class Performance { * All (but the request creation one) timestamps will be initialized with 0. */ Performance(); - Performance(Performance const&) = default; Performance& operator=(Performance const&) = default; @@ -64,45 +66,28 @@ class Performance { /** * Update object state with counters from the protocol buffer object - * - * @param workerPerformanceInfo - * counters to be carried over into an internal state + * @param workerPerformanceInfo counters to be carried over into an internal state */ void update(ProtocolPerformance const& workerPerformanceInfo); /** * Update the Controller's 'start' time - * - * @return - * the previous state of the counter + * @return the previous state of the counter */ uint64_t setUpdateStart(); /** * Update the Controller's 'finish' time - * - * @return - * the previous state of the counter + * @return the previous state of the counter */ uint64_t setUpdateFinish(); - /// Created by the Controller - uint64_t c_create_time; - - /// Started by the Controller - uint64_t c_start_time; - - /// Received by a worker service - uint64_t w_receive_time; - - /// Execution started by a worker service - uint64_t w_start_time; - - /// Execution finished by a worker service - uint64_t w_finish_time; - - /// A subscriber notified by the Controller - uint64_t c_finish_time; + uint64_t c_create_time; ///< Created by the Controller + uint64_t c_start_time; ///< Started by the Controller + uint64_t w_receive_time; ///< Received by a worker service + uint64_t w_start_time; ///< Execution started by a worker service + uint64_t w_finish_time; ///< Execution finished by a worker service + uint64_t c_finish_time; ///< A subscriber notified by the Controller }; /// Overloaded streaming operator for class Performance @@ -127,6 +112,7 @@ class WorkerPerformance { uint64_t setUpdateFinish(); std::unique_ptr info() const; + nlohmann::json toJson() const; std::atomic receive_time; ///< Received by a worker service std::atomic start_time; ///< Execution started by a worker service diff --git a/src/replica/worker/CMakeLists.txt b/src/replica/worker/CMakeLists.txt index a37868d82..c1ae9a057 100644 --- a/src/replica/worker/CMakeLists.txt +++ b/src/replica/worker/CMakeLists.txt @@ -6,9 +6,15 @@ target_sources(replica_worker PRIVATE FileServerConnection.cc WorkerDeleteRequest.cc WorkerDirectorIndexRequest.cc + WorkerEchoHttpRequest.cc WorkerEchoRequest.cc WorkerFindAllRequest.cc WorkerFindRequest.cc + WorkerHttpProcessor.cc + WorkerHttpProcessorThread.cc + WorkerHttpRequest.cc + WorkerHttpSvc.cc + WorkerHttpSvcMod.cc WorkerProcessor.cc WorkerProcessorThread.cc WorkerReplicationRequest.cc diff --git a/src/replica/worker/WorkerEchoHttpRequest.cc b/src/replica/worker/WorkerEchoHttpRequest.cc new file mode 100644 index 000000000..e2f224e12 --- /dev/null +++ b/src/replica/worker/WorkerEchoHttpRequest.cc @@ -0,0 +1,87 @@ +/* + * LSST Data Management System + * + * This product includes software developed by the + * LSST Project (http://www.lsst.org/). + * + * This program is free software: you can redistribute it and/or modify + * it under the terms of the GNU General Public License as published by + * the Free Software Foundation, either version 3 of the License, or + * (at your option) any later version. + * + * This program is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * GNU General Public License for more details. + * + * You should have received a copy of the LSST License Statement and + * the GNU General Public License along with this program. If not, + * see . + */ + +// Class header +#include "replica/worker/WorkerEchoHttpRequest.h" + +// System headers +#include + +// Qserv headers +#include "util/BlockPost.h" + +// LSST headers +#include "lsst/log/Log.h" + +using namespace std; +using json = nlohmann::json; + +namespace { + +LOG_LOGGER _log = LOG_GET("lsst.qserv.replica.WorkerEchoHttpRequest"); + +} // namespace + +namespace lsst::qserv::replica { + +shared_ptr WorkerEchoHttpRequest::create( + shared_ptr const& serviceProvider, string const& worker, + protocol::QueuedRequestHdr const& hdr, json const& req, ExpirationCallbackType const& onExpired) { + auto ptr = shared_ptr( + new WorkerEchoHttpRequest(serviceProvider, worker, hdr, req, onExpired)); + ptr->init(); + return ptr; +} + +WorkerEchoHttpRequest::WorkerEchoHttpRequest(shared_ptr const& serviceProvider, + string const& worker, protocol::QueuedRequestHdr const& hdr, + json const& req, ExpirationCallbackType const& onExpired) + : WorkerHttpRequest(serviceProvider, worker, "TEST_ECHO", hdr, req, onExpired), + _delay(req.at("delay")), + _delayLeft(req.at("delay")) {} + +void WorkerEchoHttpRequest::getResult(json& result) const { + LOGS(_log, LOG_LVL_DEBUG, context(__func__)); + replica::Lock lock(_mtx, context(__func__)); + result["data"] = req().at("data"); +} + +bool WorkerEchoHttpRequest::execute() { + LOGS(_log, LOG_LVL_DEBUG, context(__func__) << " _delay:" << _delay << " _delayLeft:" << _delayLeft); + + replica::Lock lock(_mtx, context(__func__)); + checkIfCancelling(lock, __func__); + + // Block the thread for the random number of milliseconds in the interval + // below. Then update the amount of time which is still left. + util::BlockPost blockPost(1000, 2000); + uint64_t const span = blockPost.wait(); + _delayLeft -= (span < _delayLeft) ? span : _delayLeft; + + // Done if have reached or exceeded the initial delay + if (0 == _delayLeft) { + setStatus(lock, protocol::Status::SUCCESS); + return true; + } + return false; +} + +} // namespace lsst::qserv::replica diff --git a/src/replica/worker/WorkerEchoHttpRequest.h b/src/replica/worker/WorkerEchoHttpRequest.h new file mode 100644 index 000000000..94940cc68 --- /dev/null +++ b/src/replica/worker/WorkerEchoHttpRequest.h @@ -0,0 +1,94 @@ +/* + * LSST Data Management System + * + * This product includes software developed by the + * LSST Project (http://www.lsst.org/). + * + * This program is free software: you can redistribute it and/or modify + * it under the terms of the GNU General Public License as published by + * the Free Software Foundation, either version 3 of the License, or + * (at your option) any later version. + * + * This program is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * GNU General Public License for more details. + * + * You should have received a copy of the LSST License Statement and + * the GNU General Public License along with this program. If not, + * see . + */ +#ifndef LSST_QSERV_REPLICA_WORKERECHOHTTPREQUEST_H +#define LSST_QSERV_REPLICA_WORKERECHOHTTPREQUEST_H + +// System headers +#include +#include + +// Qserv headers +#include "replica/worker/WorkerHttpRequest.h" + +// Third party headers +#include "nlohmann/json.hpp" + +// Forward declarations + +namespace lsst::qserv::replica { +class ServiceProvider; +} // namespace lsst::qserv::replica + +namespace lsst::qserv::replica::protocol { +struct QueuedRequestHdr; +} // namespace lsst::qserv::replica::protocol + +// This header declarations +namespace lsst::qserv::replica { + +/** + * Class WorkerEchoHttpRequest implements test requests within the worker servers. + * Requests of this type don't have any side effects (in terms of modifying + * any files or databases). + */ +class WorkerEchoHttpRequest : public WorkerHttpRequest { +public: + /** + * Static factory method is needed to prevent issue with the lifespan + * and memory management of instances created otherwise (as values or via + * low-level pointers). + * + * @param serviceProvider provider is needed to access the Configuration + * of a setup and for validating the input parameters + * @param worker the name of a worker. The name must match the worker which + * is going to execute the request. + * @param hdr request header (common parameters of the queued request) + * @param req the request object received from a client (request-specific parameters) + * @param onExpired request expiration callback function + * @return pointer to the created object + */ + static std::shared_ptr create( + std::shared_ptr const& serviceProvider, std::string const& worker, + protocol::QueuedRequestHdr const& hdr, nlohmann::json const& req, + ExpirationCallbackType const& onExpired); + + WorkerEchoHttpRequest() = delete; + WorkerEchoHttpRequest(WorkerEchoHttpRequest const&) = delete; + WorkerEchoHttpRequest& operator=(WorkerEchoHttpRequest const&) = delete; + + ~WorkerEchoHttpRequest() override = default; + + bool execute() override; + +protected: + WorkerEchoHttpRequest(std::shared_ptr const& serviceProvider, std::string const& worker, + protocol::QueuedRequestHdr const& hdr, nlohmann::json const& req, + ExpirationCallbackType const& onExpired); + + void getResult(nlohmann::json& result) const override; + + uint64_t _delay; ///< The amount of the initial delay + uint64_t _delayLeft; ///< The amount of the initial delay which is still left +}; + +} // namespace lsst::qserv::replica + +#endif // LSST_QSERV_REPLICA_WORKERECHOHTTPREQUEST_H diff --git a/src/replica/worker/WorkerHttpProcessor.cc b/src/replica/worker/WorkerHttpProcessor.cc new file mode 100644 index 000000000..9b1918933 --- /dev/null +++ b/src/replica/worker/WorkerHttpProcessor.cc @@ -0,0 +1,547 @@ +/* + * LSST Data Management System + * + * This product includes software developed by the + * LSST Project (http://www.lsst.org/). + * + * This program is free software: you can redistribute it and/or modify + * it under the terms of the GNU General Public License as published by + * the Free Software Foundation, either version 3 of the License, or + * (at your option) any later version. + * + * This program is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * GNU General Public License for more details. + * + * You should have received a copy of the LSST License Statement and + * the GNU General Public License along with this program. If not, + * see . + */ + +// Class header +#include "replica/worker/WorkerHttpProcessor.h" + +// System headers +#include +#include +#include + +// Qserv headers +#include "replica/config/Configuration.h" +#include "replica/mysql/DatabaseMySQL.h" +#include "replica/services/ServiceProvider.h" +#include "replica/worker/WorkerHttpProcessorThread.h" +#include "replica/worker/WorkerHttpRequest.h" +// #include "replica/worker/WorkerCreateReplicaHttpRequest.h" +// #include "replica/worker/WorkerDeleteReplicaHttpRequest.h" +// #include "replica/worker/WorkerDirectorIndexHttpRequest.h" +#include "replica/worker/WorkerEchoHttpRequest.h" +// #include "replica/worker/WorkerFindReplicaHttpRequest.h" +// #include "replica/worker/WorkerFindAllReplicasHttpRequest.h" +// #include "replica/worker/WorkerSqlHttpRequest.h" +#include "util/BlockPost.h" +#include "util/TimeUtils.h" + +// LSST headers +#include "lsst/log/Log.h" + +using namespace std; +using namespace std::placeholders; +using json = nlohmann::json; + +namespace { +LOG_LOGGER _log = LOG_GET("lsst.qserv.replica.WorkerHttpProcessor"); +} // namespace + +namespace lsst::qserv::replica { + +bool WorkerHttpProcessor::PriorityQueueType::remove(string const& id) { + auto itr = find_if(c.begin(), c.end(), + [&id](shared_ptr const& ptr) { return ptr->id() == id; }); + if (itr != c.end()) { + c.erase(itr); + make_heap(c.begin(), c.end(), comp); + return true; + } + return false; +} + +shared_ptr WorkerHttpProcessor::create( + shared_ptr const& serviceProvider, string const& worker) { + return shared_ptr(new WorkerHttpProcessor(serviceProvider, worker)); +} + +WorkerHttpProcessor::WorkerHttpProcessor(shared_ptr const& serviceProvider, + string const& worker) + : _serviceProvider(serviceProvider), + _worker(worker), + _connectionPool(database::mysql::ConnectionPool::create( + Configuration::qservWorkerDbParams(), + serviceProvider->config()->get("database", "services-pool-size"))), + _state(protocol::ServiceState::SUSPENDED), + _startTime(util::TimeUtils::now()) {} + +void WorkerHttpProcessor::run() { + LOGS(_log, LOG_LVL_DEBUG, _context(__func__)); + replica::Lock lock(_mtx, _context(__func__)); + + if (_state == protocol::ServiceState::SUSPENDED) { + size_t const numThreads = + _serviceProvider->config()->get("worker", "num-svc-processing-threads"); + if (not numThreads) { + throw out_of_range(_classMethodContext(__func__) + + "invalid configuration parameter for the number of processing threads. " + "The value of the parameter must be greater than 0"); + } + + // Create threads if needed + if (_threads.empty()) { + auto const self = shared_from_this(); + for (size_t i = 0; i < numThreads; ++i) { + _threads.push_back(WorkerHttpProcessorThread::create(self)); + } + } + + // Tell each thread to run + for (auto&& t : _threads) { + t->run(); + } + _state = protocol::ServiceState::RUNNING; + } +} + +void WorkerHttpProcessor::stop() { + LOGS(_log, LOG_LVL_DEBUG, _context(__func__)); + replica::Lock lock(_mtx, _context(__func__)); + + if (_state == protocol::ServiceState::RUNNING) { + // Tell each thread to stop. + for (auto&& t : _threads) { + t->stop(); + } + + // Begin transitioning to the final state via this intermediate one. + // The transition will finish asynchronous when all threads will report + // desired changes in their states. + _state = protocol::ServiceState::SUSPEND_IN_PROGRESS; + } +} + +void WorkerHttpProcessor::drain() { + LOGS(_log, LOG_LVL_DEBUG, _context(__func__)); + replica::Lock lock(_mtx, _context(__func__)); + + // Collect identifiers of requests to be affected by the operation + list ids; + for (auto&& ptr : _newRequests) ids.push_back(ptr->id()); + for (auto&& entry : _inProgressRequests) ids.push_back(entry.first); + for (auto&& id : ids) _stopRequestImpl(lock, id); +} + +void WorkerHttpProcessor::reconfig() { + LOGS(_log, LOG_LVL_DEBUG, _context(__func__)); + replica::Lock lock(_mtx, _context(__func__)); + _serviceProvider->config()->reload(); +} + +json WorkerHttpProcessor::createReplica(protocol::QueuedRequestHdr const& hdr, json const& req) { + LOGS(_log, LOG_LVL_DEBUG, _context(__func__) << " id: " << hdr.id); + // return _submit(replica::Lock(_mtx, context), context, hdr, req); + return json::object(); +} + +json WorkerHttpProcessor::deleteReplica(protocol::QueuedRequestHdr const& hdr, json const& req) { + LOGS(_log, LOG_LVL_DEBUG, _context(__func__) << " id: " << hdr.id); + // return _submit(replica::Lock(_mtx, context), context, hdr, req); + return json::object(); +} + +json WorkerHttpProcessor::findReplica(protocol::QueuedRequestHdr const& hdr, json const& req) { + LOGS(_log, LOG_LVL_DEBUG, _context(__func__) << " id: " << hdr.id); + // return _submit(replica::Lock(_mtx, context), context, hdr, req); + return json::object(); +} + +json WorkerHttpProcessor::findAllReplicas(protocol::QueuedRequestHdr const& hdr, json const& req) { + LOGS(_log, LOG_LVL_DEBUG, _context(__func__) << " id: " << hdr.id); + // return _submit(replica::Lock(_mtx, context), context, hdr, req); + return json::object(); +} + +json WorkerHttpProcessor::echo(protocol::QueuedRequestHdr const& hdr, json const& req) { + LOGS(_log, LOG_LVL_DEBUG, _context(__func__) << " id: " << hdr.id); + string const context = _context(__func__); + return _submit(replica::Lock(_mtx, context), context, hdr, req); +} + +json WorkerHttpProcessor::sql(protocol::QueuedRequestHdr const& hdr, json const& req) { + LOGS(_log, LOG_LVL_DEBUG, _context(__func__) << " id: " << hdr.id); + // return _submit(replica::Lock(_mtx, context), context, hdr, req); + return json::object(); +} + +json WorkerHttpProcessor::index(protocol::QueuedRequestHdr const& hdr, json const& req) { + LOGS(_log, LOG_LVL_DEBUG, _context(__func__) << " id: " << hdr.id); + // return _submit(replica::Lock(_mtx, context), context, hdr, req); + return json::object(); +} + +json WorkerHttpProcessor::requestStatus(string const& id) { + LOGS(_log, LOG_LVL_DEBUG, _context(__func__) << " id: " << id); + replica::Lock lock(_mtx, _context(__func__)); + + // Still waiting in the queue? + shared_ptr targetRequestPtr; + for (auto ptr : _newRequests) { + if (ptr->id() == id) { + targetRequestPtr = ptr; + break; + } + } + if (targetRequestPtr == nullptr) { + // Is it already being processed? + auto itrInProgress = _inProgressRequests.find(id); + if (itrInProgress != _inProgressRequests.end()) { + targetRequestPtr = itrInProgress->second; + } + if (targetRequestPtr == nullptr) { + // Has it finished? + auto itrFinished = _finishedRequests.find(id); + if (itrFinished != _finishedRequests.end()) { + targetRequestPtr = itrFinished->second; + } + // No such request? + if (targetRequestPtr == nullptr) { + return json::object({ + {"status", protocol::Status::BAD}, + {"status_ext", protocol::StatusExt::INVALID_ID}, + }); + } + } + } + return targetRequestPtr->toJson(); +} + +json WorkerHttpProcessor::stopRequest(string const& id) { + LOGS(_log, LOG_LVL_DEBUG, _context(__func__) << " id: " << id); + replica::Lock lock(_mtx, _context(__func__)); + json response = json::object(); + auto const request = _stopRequestImpl(lock, id); + if (request == nullptr) { + response["status"] = protocol::Status::BAD; + response["status_ext"] = protocol::StatusExt::INVALID_ID; + } else { + response = request->toJson(); + } + return response; +} + +json WorkerHttpProcessor::trackRequest(string const& id) { + LOGS(_log, LOG_LVL_DEBUG, _context(__func__) << " id: " << id); + replica::Lock lock(_mtx, _context(__func__)); + json response = json::object(); + auto const request = _trackRequestImpl(lock, id); + if (request == nullptr) { + response["status"] = protocol::Status::BAD; + response["status_ext"] = protocol::StatusExt::INVALID_ID; + } else { + bool const includeResultIfFinished = true; + response = request->toJson(includeResultIfFinished); + } + return response; +} + +bool WorkerHttpProcessor::disposeRequest(string const& id) { + LOGS(_log, LOG_LVL_DEBUG, _context(__func__) << " id: " << id); + replica::Lock lock(_mtx, _context(__func__)); + + // Note that only the finished requests are allowed to be disposed. + if (auto itr = _finishedRequests.find(id); itr != _finishedRequests.end()) { + itr->second->dispose(); + _finishedRequests.erase(itr); + return true; + } + return false; +} + +size_t WorkerHttpProcessor::numNewRequests() const { + replica::Lock lock(_mtx, _context(__func__)); + return _newRequests.size(); +} + +size_t WorkerHttpProcessor::numInProgressRequests() const { + replica::Lock lock(_mtx, _context(__func__)); + return _inProgressRequests.size(); +} + +size_t WorkerHttpProcessor::numFinishedRequests() const { + replica::Lock lock(_mtx, _context(__func__)); + return _finishedRequests.size(); +} + +json WorkerHttpProcessor::toJson(protocol::Status status, bool includeRequests) { + LOGS(_log, LOG_LVL_DEBUG, _context(__func__)); + replica::Lock lock(_mtx, _context(__func__)); + + json response; + response["status"] = status; + response["status_ext"] = protocol::StatusExt::NONE; + response["service_state"] = state(); + response["num_new_requests"] = _newRequests.size(); + response["num_in_progress_requests"] = _inProgressRequests.size(); + response["num_finished_requests"] = _finishedRequests.size(); + response["new_requests"] = json::array(); + response["in_progress_requests"] = json::array(); + response["finished_requests"] = json::array(); + + if (includeRequests) { + for (auto const& request : _newRequests) { + response["new_requests"].push_back(request->toJson()); + } + for (auto const& entry : _inProgressRequests) { + response["in_progress_requests"].push_back(entry.second->toJson()); + } + for (auto const& entry : _finishedRequests) { + response["finished_requests"].push_back(entry.second->toJson()); + } + } + return response; +} + +string WorkerHttpProcessor::_classMethodContext(string const& func) { return "WorkerHttpProcessor::" + func; } + +void WorkerHttpProcessor::_logError(string const& context, string const& message) const { + LOGS(_log, LOG_LVL_ERROR, context << " " << message); +} + +shared_ptr WorkerHttpProcessor::_stopRequestImpl(replica::Lock const& lock, + string const& id) { + LOGS(_log, LOG_LVL_DEBUG, _context(__func__) << " id: " << id); + + // Still waiting in the queue? + // + // ATTENTION: the loop variable is a copy of (not a reference to) a shared + // pointer to allow removing (if needed) the corresponding entry from the + // input collection while retaining a valid copy of the pointer to be placed + // into the next stage collection. + + for (auto ptr : _newRequests) { + if (ptr->id() == id) { + // Cancel it and move it into the final queue in case if a client + // won't be able to receive the desired status of the request due to + // a protocol failure, etc. + ptr->cancel(); + switch (ptr->status()) { + case protocol::Status::CANCELLED: { + _newRequests.remove(id); + _finishedRequests[ptr->id()] = ptr; + return ptr; + } + default: + throw logic_error(_classMethodContext(__func__) + " unexpected request status " + + protocol::toString(ptr->status()) + " in new requests"); + } + } + } + + // Is it already being processed? + auto itrInProgress = _inProgressRequests.find(id); + if (itrInProgress != _inProgressRequests.end()) { + auto ptr = itrInProgress->second; + // Tell the request to begin the cancelling protocol. The protocol + // will take care of moving the request into the final queue when + // the cancellation will finish. + // + // At the meant time we just notify the client about the cancellation status + // of the request and let it come back later to check the updated status. + ptr->cancel(); + switch (ptr->status()) { + // These are the most typical states for request in this queue + case protocol::Status::CANCELLED: + case protocol::Status::IS_CANCELLING: + + // The following two states are also allowed here because + // in-progress requests are still allowed to progress to the completed + // states before reporting their new state via method: + // WorkerHttpProcessor::_processingFinished() + // Sometimes, the request just can't finish this in time due to + // replica::Lock lock(_mtx) held by the current method. We shouldn't worry + // about this situation here. The request will be moved into the next + // queue as soon as replica::Lock lock(_mtx) will be released. + case protocol::Status::SUCCESS: + case protocol::Status::FAILED: + return ptr; + default: + throw logic_error(_classMethodContext(__func__) + " unexpected request status " + + protocol::toString(ptr->status()) + " in in-progress requests"); + } + } + + // Has it finished? + auto itrFinished = _finishedRequests.find(id); + if (itrFinished != _finishedRequests.end()) { + auto ptr = itrFinished->second; + // There is nothing else we can do here other than just + // reporting the completion status of the request. It's up to a client + // to figure out what to do about this situation. + switch (ptr->status()) { + case protocol::Status::CANCELLED: + case protocol::Status::SUCCESS: + case protocol::Status::FAILED: + return ptr; + default: + throw logic_error(_classMethodContext(__func__) + " unexpected request status " + + protocol::toString(ptr->status()) + " in finished requests"); + } + } + + // No request found! + return nullptr; +} + +shared_ptr WorkerHttpProcessor::_trackRequestImpl(replica::Lock const& lock, + string const& id) { + LOGS(_log, LOG_LVL_DEBUG, _context(__func__) << " id: " << id); + + // Still waiting in the queue? + for (auto&& ptr : _newRequests) { + if (ptr->id() == id) { + switch (ptr->status()) { + // This state requirement is strict for the non-active requests + case protocol::Status::CREATED: + return ptr; + default: + throw logic_error(_classMethodContext(__func__) + " unexpected request status " + + protocol::toString(ptr->status()) + " in new requests"); + } + } + } + + // Is it already being processed? + auto itrInProgress = _inProgressRequests.find(id); + if (itrInProgress != _inProgressRequests.end()) { + auto ptr = itrInProgress->second; + switch (ptr->status()) { + // These are the most typical states for request in this queue + case protocol::Status::IS_CANCELLING: + case protocol::Status::IN_PROGRESS: + + // The following three states are also allowed here because + // in-progress requests are still allowed to progress to the completed + // states before reporting their new state via method: + // WorkerHttpProcessor::_processingFinished() + // Sometimes, the request just can't finish this in time due to + // replica::Lock lock(_mtx) held by the current method. We shouldn't worry + // about this situation here. The request will be moved into the next + // queue as soon as replica::Lock lock(_mtx) will be released. + case protocol::Status::CANCELLED: + case protocol::Status::SUCCESS: + case protocol::Status::FAILED: + return ptr; + default: + throw logic_error(_classMethodContext(__func__) + " unexpected request status " + + protocol::toString(ptr->status()) + " in in-progress requests"); + } + } + + // Has it finished? + auto itrFinished = _finishedRequests.find(id); + if (itrFinished != _finishedRequests.end()) { + auto ptr = itrFinished->second; + switch (ptr->status()) { + // This state requirement is strict for the completed requests + case protocol::Status::CANCELLED: + case protocol::Status::SUCCESS: + case protocol::Status::FAILED: + return ptr; + default: + throw logic_error(_classMethodContext(__func__) + " unexpected request status " + + protocol::toString(ptr->status()) + " in finished requests"); + } + } + + // No request found! + return nullptr; +} + +shared_ptr WorkerHttpProcessor::_fetchNextForProcessing( + shared_ptr const& processorThread, unsigned int timeoutMilliseconds) { + LOGS(_log, LOG_LVL_TRACE, + _context(__func__) << " thread: " << processorThread->id() << " timeout: " << timeoutMilliseconds); + + // For generating random intervals within the maximum range of seconds + // requested by a client. + // + // TODO: Re-implement this loop to use a condition variable instead. + // This will improve the performance of the processor which is limited + // by the half-latency of the wait interval. + util::BlockPost blockPost(0, min(10U, timeoutMilliseconds)); + + unsigned int totalElapsedTime = 0; + while (totalElapsedTime < timeoutMilliseconds) { + // IMPORTANT: make sure no wait is happening within the same + // scope where the thread safe block is defined. Otherwise + // the queue will be locked for all threads for the duration of + // the wait. + { + replica::Lock lock(_mtx, _context(__func__)); + if (!_newRequests.empty()) { + shared_ptr request = _newRequests.top(); + _newRequests.pop(); + request->start(); + _inProgressRequests[request->id()] = request; + return request; + } + } + totalElapsedTime += blockPost.wait(); + } + + // Return null pointer since noting has been found within the specified + // timeout. + return nullptr; +} + +void WorkerHttpProcessor::_processingRefused(shared_ptr const& request) { + LOGS(_log, LOG_LVL_DEBUG, _context(__func__) << " id: " << request->id()); + replica::Lock lock(_mtx, _context(__func__)); + + // Note that disposed requests won't be found in any queue. + auto itr = _inProgressRequests.find(request->id()); + if (itr != _inProgressRequests.end()) { + // Update request's state before moving it back into + // the input queue. + itr->second->stop(); + _newRequests.push(itr->second); + _inProgressRequests.erase(itr); + } +} + +void WorkerHttpProcessor::_processingFinished(shared_ptr const& request) { + LOGS(_log, LOG_LVL_DEBUG, + _context(__func__) << " id: " << request->id() + << " status: " << protocol::toString(request->status())); + replica::Lock lock(_mtx, _context(__func__)); + + // Note that disposed requests won't be found in any queue. + auto itr = _inProgressRequests.find(request->id()); + if (itr != _inProgressRequests.end()) { + _finishedRequests[itr->first] = itr->second; + _inProgressRequests.erase(itr); + } +} + +void WorkerHttpProcessor::_processorThreadStopped( + shared_ptr const& processorThread) { + LOGS(_log, LOG_LVL_DEBUG, _context(__func__) << " thread: " << processorThread->id()); + replica::Lock lock(_mtx, _context(__func__)); + if (_state == protocol::ServiceState::SUSPEND_IN_PROGRESS) { + // Complete state transition if all threads are stopped + for (auto&& t : _threads) { + if (t->isRunning()) return; + } + _state = protocol::ServiceState::SUSPENDED; + } +} + +} // namespace lsst::qserv::replica diff --git a/src/replica/worker/WorkerHttpProcessor.h b/src/replica/worker/WorkerHttpProcessor.h new file mode 100644 index 000000000..d05e495a4 --- /dev/null +++ b/src/replica/worker/WorkerHttpProcessor.h @@ -0,0 +1,365 @@ +/* + * LSST Data Management System + * + * This product includes software developed by the + * LSST Project (http://www.lsst.org/). + * + * This program is free software: you can redistribute it and/or modify + * it under the terms of the GNU General Public License as published by + * the Free Software Foundation, either version 3 of the License, or + * (at your option) any later version. + * + * This program is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * GNU General Public License for more details. + * + * You should have received a copy of the LSST License Statement and + * the GNU General Public License along with this program. If not, + * see . + */ +#ifndef LSST_QSERV_REPLICA_WORKERHTTPPROCESSOR_H +#define LSST_QSERV_REPLICA_WORKERHTTPPROCESSOR_H + +// System headers +#include +#include +#include +#include +#include +#include +#include +#include + +// Qserv headers +#include "replica/proto/Protocol.h" +#include "replica/util/Mutex.h" +#include "replica/worker/WorkerHttpRequest.h" + +// Third party headers +#include "nlohmann/json.hpp" + +// Forward declarations + +namespace lsst::qserv::replica { +class ServiceProvider; +class WorkerHttpProcessorThread; +} // namespace lsst::qserv::replica + +namespace lsst::qserv::replica::database::mysql { +class ConnectionPool; +} // namespace lsst::qserv::replica::database::mysql + +// This header declarations +namespace lsst::qserv::replica { + +/** + * Class WorkerHttpProcessor is a front-end interface for processing + * requests from remote clients within worker-side services. + */ +class WorkerHttpProcessor : public std::enable_shared_from_this { +public: + // The thread-based processor class is allowed to access the internal API + friend class WorkerHttpProcessorThread; + + /** + * Structure PriorityQueueType extends the standard priority queue for pointers + * to the new (unprocessed) requests. + * + * Its design relies upon the inheritance to get access to the protected + * data members 'c' representing the internal container of the base queue + * in order to implement the iterator protocol. + */ + struct PriorityQueueType + : std::priority_queue, + std::vector>, WorkerHttpRequestCompare> { + /// @return iterator to the beginning of the container + decltype(c.begin()) begin() { return c.begin(); } + + /// @return iterator to the end of the container + decltype(c.end()) end() { return c.end(); } + + /** + * Remove a request from the queue by its identifier + * @param id an identifier of a request + * @return 'true' if the object was actually removed + */ + bool remove(std::string const& id); + }; + + /** + * The factory method for objects of the class + * + * @param serviceProvider provider is needed to access the Configuration of + * a setup in order to get a number of the processing threads to be launched + * by the processor. + * @param worker the name of a worker + * @return a pointer to the created object + */ + static std::shared_ptr create( + std::shared_ptr const& serviceProvider, std::string const& worker); + + WorkerHttpProcessor() = delete; + WorkerHttpProcessor(WorkerHttpProcessor const&) = delete; + WorkerHttpProcessor& operator=(WorkerHttpProcessor const&) = delete; + + ~WorkerHttpProcessor() = default; + + /// @return the state of the processor + protocol::ServiceState state() const { return _state; } + + /// Begin processing requests + void run(); + + /// Stop processing all requests, and stop all threads + void stop(); + + /// Drain (cancel) all queued and in-progress requests + void drain(); + + /// Reload Configuration + void reconfig(); + + /** + * Enqueue the replica creation request for processing + * @param hdr request header (common parameters of the queued request) + * @param req the request object received from a client (request-specific parameters) + * @return the response object to be sent back to a client + */ + nlohmann::json createReplica(protocol::QueuedRequestHdr const& hdr, nlohmann::json const& req); + + /** + * Enqueue the replica deletion request for processing + * @param hdr request header (common parameters of the queued request) + * @param req the request object received from a client (request-specific parameters) + * @return the response object to be sent back to a client + */ + nlohmann::json deleteReplica(protocol::QueuedRequestHdr const& hdr, nlohmann::json const& req); + + /** + * Enqueue the replica lookup request for processing + * @param hdr request header (common parameters of the queued request) + * @param req the request object received from a client (request-specific parameters) + * @return the response object to be sent back to a client + */ + nlohmann::json findReplica(protocol::QueuedRequestHdr const& hdr, nlohmann::json const& req); + + /** + * Enqueue the multi-replica lookup request for processing + * @param hdr request header (common parameters of the queued request) + * @param req the request object received from a client (request-specific parameters) + * @return the response object to be sent back to a client + */ + nlohmann::json findAllReplicas(protocol::QueuedRequestHdr const& hdr, nlohmann::json const& req); + + /** + * Enqueue the worker-side testing request for processing + * @param hdr request header (common parameters of the queued request) + * @param req the request object received from a client (request-specific parameters) + * @return the response object to be sent back to a client + */ + nlohmann::json echo(protocol::QueuedRequestHdr const& hdr, nlohmann::json const& req); + + /** + * Enqueue a request for querying the worker database + * @param hdr request header (common parameters of the queued request) + * @param req the request object received from a client (request-specific parameters) + * @return the response object to be sent back to a client + */ + nlohmann::json sql(protocol::QueuedRequestHdr const& hdr, nlohmann::json const& req); + + /** + * Enqueue a request for extracting the "director" index data from + * the director tables. + * @param hdr request header (common parameters of the queued request) + * @param req the request object received from a client (request-specific parameters) + * @return the response object to be sent back to a client + */ + nlohmann::json index(protocol::QueuedRequestHdr const& hdr, nlohmann::json const& req); + + /** + * Get a status of the request + * @param id an identifier of a request affected by the operation + * @return the response object to be sent back to a client + */ + nlohmann::json requestStatus(std::string const& id); + + /** + * Dequeue replication request + * @note If the request is not being processed yet then it will be simply removed + * from the ready-to-be-processed queue. If it's being processed an attempt + * to cancel processing will be made. If it has already processed this will + * be reported. + * @param id an identifier of a request affected by the operation + * @return the response object to be sent back to a client + */ + nlohmann::json stopRequest(std::string const& id); + + /** + * Return the tracking info on the on-going request + * @param id an identifier of a request affected by the operation + * @return the response object to be sent back to a client + */ + nlohmann::json trackRequest(std::string const& id); + + /** + * Find the request in any queue, and "garbage collect" it to release resources + * associated with the request. If the request is still in the "in-progress" + * state then it will be "drained" before disposing. If the request isn't found + * in any queue then nothing will happen (no exception thrown, no side effects). + * + * @param id an identifier of a request affected by the operation + * @return 'true' if the request was found and actually removed from any queue + */ + bool disposeRequest(std::string const& id); + + size_t numNewRequests() const; + size_t numInProgressRequests() const; + size_t numFinishedRequests() const; + + /** + * Capture the processor's state and counters. + * @param status desired status to set in the response objet + * @param includeRequests (optional) flag to return detailed info on all known requests + * @return the response object to be sent back to a client + */ + nlohmann::json toJson(protocol::Status status, bool includeRequests = false); + +private: + WorkerHttpProcessor(std::shared_ptr const& serviceProvider, std::string const& worker); + + static std::string _classMethodContext(std::string const& func); + + /** + * Submit a request for processing + * @param lock a lock on _mtx to be acquired before calling this method + * @param context the logging context (including the name of a function/method) + * @param hdr request header (common parameters of the queued request) + * @param req the request object received from a client (request-specific parameters) + * @return the response object to be sent back to a client + */ + template + nlohmann::json _submit(replica::Lock const& lock, std::string const& context, + protocol::QueuedRequestHdr const& hdr, nlohmann::json const& req) { + try { + auto const ptr = REQUEST_TYPE::create( + _serviceProvider, _worker, hdr, req, + [self = shared_from_this()](std::string const& id) { self->disposeRequest(id); }); + _newRequests.push(ptr); + return ptr->toJson(); + } catch (std::exception const& ec) { + _logError(context, ec.what()); + return nlohmann::json::object({ + {"status", protocol::Status::BAD}, + {"status_ext", protocol::StatusExt::INVALID_PARAM}, + }); + } + } + + /** + * Log the error message. + * @param context the logging context (including the name of a function/method) + * @param message the error message to be reported + */ + void _logError(std::string const& context, std::string const& message) const; + + /** + * Return the next request which is ready to be processed + * and if then one found assign it to the specified thread. The request + * will be removed from the ready-to-be-processed queue. + * + * If the one is available within the specified timeout then such request + * will be moved into the in-progress queue, assigned to the processor thread + * and returned to a caller. Otherwise an empty pointer (pointing to nullptr) + * will be returned. + * + * This method is supposed to be called by one of the processing threads + * when it becomes available. + * + * @note this method will block for a duration of time not exceeding + * the client-specified timeout unless it's set to 0. In the later + * case the method will block indefinitely. + * @param processorThread reference to a thread which fetches the next request + * @param timeoutMilliseconds (optional) amount of time to wait before to finish if + * no suitable requests are available for processing + */ + std::shared_ptr _fetchNextForProcessing( + std::shared_ptr const& processorThread, + unsigned int timeoutMilliseconds = 0); + + /** + * Implement the operation for the specified identifier if such request + * is still known to the Processor. Return a reference to the request object + * whose state will be properly updated. + * @param lock а lock on _mtx to be acquired before calling this method + * @param id an identifier of a request + * @return the request object (if found) or nullptr otherwise + */ + std::shared_ptr _stopRequestImpl(replica::Lock const& lock, std::string const& id); + + /** + * Find and return a reference to the request object. + * @param lock а lock on _mtx to be acquired before calling this method + * @param id an identifier of a request + * @return the request object (if found) or nullptr otherwise + */ + std::shared_ptr _trackRequestImpl(replica::Lock const& lock, std::string const& id); + + /** + * Report a decision not to process a request + * + * This method is supposed to be called by one of the processing threads + * after it fetches the next ready-to-process request and then decided + * not to proceed with processing. Normally this should happen when + * the thread was asked to stop. In that case the request will be put + * back into the ready-to-be processed request and be picked up later + * by some other thread. + * + * @param request a pointer to the request + */ + void _processingRefused(std::shared_ptr const& request); + + /** + * Report a request which has been processed or cancelled. + * + * The method is called by a thread which was processing the request. + * The request will be moved into the corresponding queue. A proper + * completion status is expected be stored within the request. + * + * @param request a pointer to the request + */ + void _processingFinished(std::shared_ptr const& request); + + /** + * For threads reporting their completion + * + * This method is used by threads to report a change in their state. + * It's meant to be used during the gradual and asynchronous state transition + * of this processor from the combined State::STATE_IS_STOPPING to + * State::STATE_IS_STOPPED. The later is achieved when all threads are stopped. + * + * @param processorThread reference to the processing thread which finished + */ + void _processorThreadStopped(std::shared_ptr const& processorThread); + + std::string _context(std::string const& func = std::string()) const { return "PROCESSOR " + func; } + + std::shared_ptr const _serviceProvider; + std::string const _worker; + std::shared_ptr const _connectionPool; + + protocol::ServiceState _state; + uint64_t _startTime; /// When the processor started (milliseconds since UNIX Epoch) + + std::vector> _threads; + + mutable replica::Mutex _mtx; /// Mutex guarding the queues + + PriorityQueueType _newRequests; + std::map> _inProgressRequests; + std::map> _finishedRequests; +}; + +} // namespace lsst::qserv::replica + +#endif // LSST_QSERV_REPLICA_WORKERHTTPPROCESSOR_H diff --git a/src/replica/worker/WorkerHttpProcessorThread.cc b/src/replica/worker/WorkerHttpProcessorThread.cc new file mode 100644 index 000000000..e163b79d7 --- /dev/null +++ b/src/replica/worker/WorkerHttpProcessorThread.cc @@ -0,0 +1,122 @@ +/* + * LSST Data Management System + * + * This product includes software developed by the + * LSST Project (http://www.lsst.org/). + * + * This program is free software: you can redistribute it and/or modify + * it under the terms of the GNU General Public License as published by + * the Free Software Foundation, either version 3 of the License, or + * (at your option) any later version. + * + * This program is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * GNU General Public License for more details. + * + * You should have received a copy of the LSST License Statement and + * the GNU General Public License along with this program. If not, + * see . + */ + +// Class header +#include "replica/worker/WorkerHttpProcessorThread.h" + +// System headers +#include + +// Qserv headers +#include "replica/proto/Protocol.h" +#include "replica/worker/WorkerHttpProcessor.h" +#include "replica/worker/WorkerHttpRequest.h" + +// LSST headers +#include "lsst/log/Log.h" + +using namespace std; + +namespace { + +LOG_LOGGER _log = LOG_GET("lsst.qserv.replica.WorkerHttpProcessorThread"); + +} // namespace + +namespace lsst::qserv::replica { + +shared_ptr WorkerHttpProcessorThread::create( + shared_ptr const& processor) { + static unsigned int id = 0; + return shared_ptr(new WorkerHttpProcessorThread(processor, id++)); +} + +WorkerHttpProcessorThread::WorkerHttpProcessorThread(shared_ptr const& processor, + unsigned int id) + : _processor(processor), _id(id), _stop(false) {} + +bool WorkerHttpProcessorThread::isRunning() const { return _thread != nullptr; } + +void WorkerHttpProcessorThread::run() { + if (isRunning()) return; + + _thread = make_unique([self = shared_from_this()]() { + LOGS(_log, LOG_LVL_DEBUG, self->context() << "start"); + while (not self->_stop) { + // Get the next request to process if any. This operation will block + // until either the next request is available (returned a valid pointer) + // or the specified timeout expires. In either case this thread has a chance + // to re-evaluate the stopping condition. + auto const request = self->_processor->_fetchNextForProcessing(self, 1000); + if (self->_stop) { + if (request) self->_processor->_processingRefused(request); + continue; + } + if (request) { + LOGS(_log, LOG_LVL_DEBUG, + self->context() << "begin processing" + << " id: " << request->id()); + bool finished = false; // just to report the request completion + try { + while (not(finished = request->execute())) { + if (self->_stop) { + LOGS(_log, LOG_LVL_DEBUG, + self->context() << "rollback processing" + << " id: " << request->id()); + request->rollback(); + self->_processor->_processingRefused(request); + break; + } + } + } catch (WorkerHttpRequestCancelled const& ex) { + LOGS(_log, LOG_LVL_DEBUG, + self->context() << "cancel processing" + << " id: " << request->id()); + self->_processor->_processingFinished(request); + } + if (finished) { + LOGS(_log, LOG_LVL_DEBUG, + self->context() << "finish processing" + << " id: " << request->id() + << " status: " << protocol::toString(request->status())); + self->_processor->_processingFinished(request); + } + } + } + LOGS(_log, LOG_LVL_DEBUG, self->context() << "stop"); + + self->_stopped(); + }); + _thread->detach(); +} + +void WorkerHttpProcessorThread::stop() { + if (not isRunning()) return; + _stop = true; +} + +void WorkerHttpProcessorThread::_stopped() { + _stop = false; + _thread.reset(nullptr); + _processor->_processorThreadStopped(shared_from_this()); +} + +} // namespace lsst::qserv::replica diff --git a/src/replica/worker/WorkerHttpProcessorThread.h b/src/replica/worker/WorkerHttpProcessorThread.h new file mode 100644 index 000000000..388a30faf --- /dev/null +++ b/src/replica/worker/WorkerHttpProcessorThread.h @@ -0,0 +1,113 @@ +/* + * LSST Data Management System + * + * This product includes software developed by the + * LSST Project (http://www.lsst.org/). + * + * This program is free software: you can redistribute it and/or modify + * it under the terms of the GNU General Public License as published by + * the Free Software Foundation, either version 3 of the License, or + * (at your option) any later version. + * + * This program is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * GNU General Public License for more details. + * + * You should have received a copy of the LSST License Statement and + * the GNU General Public License along with this program. If not, + * see . + */ +#ifndef LSST_QSERV_REPLICA_WORKERHTTPPROCESSORTHREAD_H +#define LSST_QSERV_REPLICA_WORKERHTTPPROCESSORTHREAD_H + +// System headers +#include +#include +#include +#include + +// Forward declarations +namespace lsst::qserv::replica { +class WorkerHttpProcessor; +} // namespace lsst::qserv::replica + +// This header declarations +namespace lsst::qserv::replica { + +/** + * Class WorkerHttpProcessorThread is a thread-based request processing engine + * for replication requests within worker-side services. + */ +class WorkerHttpProcessorThread : public std::enable_shared_from_this { +public: + /** + * Static factory method is needed to prevent issue with the lifespan + * and memory management of instances created otherwise (as values or via + * low-level pointers). + * + * @param processor A pointer to the processor which launched this thread. This pointer + * will be used for making call backs to the processor on the completed or rejected requests. + * @return a pointer to the created object + */ + static std::shared_ptr create( + std::shared_ptr const& processor); + + WorkerHttpProcessorThread() = delete; + WorkerHttpProcessorThread(WorkerHttpProcessorThread const&) = delete; + WorkerHttpProcessorThread& operator=(WorkerHttpProcessorThread const&) = delete; + + ~WorkerHttpProcessorThread() = default; + + /// @return identifier of this thread object + unsigned int id() const { return _id; } + + /// @return 'true' if the processing thread is still running + bool isRunning() const; + + /** + * Create and run the thread (if none is still running) fetching + * and processing requests until method stop() is called. + */ + void run(); + + /** + * Tell the running thread to abort processing the current + * request (if any), put that request back into the input queue, + * stop fetching new requests and finish. The thread can be resumed + * later by calling method run(). + * + * @note This is an asynchronous operation. + */ + void stop(); + + /// @return context string for logs + std::string context() const { return "THREAD: " + std::to_string(_id) + " "; } + +private: + /// @see WorkerHttpProcessorThread::create() + WorkerHttpProcessorThread(std::shared_ptr const& processor, unsigned int id); + + /** + * Event handler called by the thread when it's about to stop + */ + void _stopped(); + + // Input parameters + + std::shared_ptr const _processor; + + /// The identifier of this thread object + unsigned int const _id; + + /// The processing thread is created on demand when calling method run() + std::unique_ptr _thread; + + /// The flag to be raised to tell the running thread to stop. + /// The thread will reset this flag when it finishes. + std::atomic _stop; +}; + +} // namespace lsst::qserv::replica + +#endif // LSST_QSERV_REPLICA_WORKERHTTPPROCESSORTHREAD_H diff --git a/src/replica/worker/WorkerHttpRequest.cc b/src/replica/worker/WorkerHttpRequest.cc new file mode 100644 index 000000000..d2044cb92 --- /dev/null +++ b/src/replica/worker/WorkerHttpRequest.cc @@ -0,0 +1,272 @@ +/* + * LSST Data Management System + * + * This product includes software developed by the + * LSST Project (http://www.lsst.org/). + * + * This program is free software: you can redistribute it and/or modify + * it under the terms of the GNU General Public License as published by + * the Free Software Foundation, either version 3 of the License, or + * (at your option) any later version. + * + * This program is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * GNU General Public License for more details. + * + * You should have received a copy of the LSST License Statement and + * the GNU General Public License along with this program. If not, + * see . + */ + +// Class header +#include "replica/worker/WorkerHttpRequest.h" + +// System headers +#include + +// Third party headers +#include "boost/date_time/posix_time/posix_time.hpp" + +// Qserv headers +#include "replica/config/Configuration.h" +#include "replica/services/ServiceProvider.h" + +// LSST headers +#include "lsst/log/Log.h" + +using namespace std; +using namespace std::placeholders; +using json = nlohmann::json; + +namespace { +LOG_LOGGER _log = LOG_GET("lsst.qserv.replica.WorkerHttpRequest"); +} // namespace + +namespace lsst::qserv::replica { + +replica::Mutex WorkerHttpRequest::_mtxDataFolderOperations; + +atomic WorkerHttpRequest::_numInstances{0}; + +WorkerHttpRequest::WorkerHttpRequest(shared_ptr const& serviceProvider, string const& worker, + string const& type, protocol::QueuedRequestHdr const& hdr, + json const& req, ExpirationCallbackType const& onExpired) + : _serviceProvider(serviceProvider), + _worker(worker), + _type(type), + _hdr(hdr), + _req(req), + _onExpired(onExpired), + _expirationTimeoutSec(hdr.timeout == 0 ? serviceProvider->config()->get( + "controller", "request-timeout-sec") + : hdr.timeout), + _expirationTimer(serviceProvider->io_service()), + _status(protocol::Status::CREATED), + _extendedStatus(protocol::StatusExt::NONE), + _performance() { + _numInstances++; + LOGS(_log, LOG_LVL_TRACE, context(__func__) << " numInstances: " << _numInstances); +} + +WorkerHttpRequest::~WorkerHttpRequest() { + _numInstances--; + LOGS(_log, LOG_LVL_TRACE, context(__func__) << " numInstances: " << _numInstances); + dispose(); +} + +void WorkerHttpRequest::checkIfCancelling(replica::Lock const& lock, string const& func) { + switch (status()) { + case protocol::Status::IN_PROGRESS: + break; + case protocol::Status::IS_CANCELLING: + setStatus(lock, protocol::Status::CANCELLED); + throw WorkerHttpRequestCancelled(); + default: + throw logic_error(context(func) + + " not allowed while in status: " + protocol::toString(status())); + } +} + +WorkerHttpRequest::ErrorContext WorkerHttpRequest::reportErrorIf(bool errorCondition, + protocol::StatusExt extendedStatus, + string const& errorMsg) { + WorkerHttpRequest::ErrorContext errorContext; + if (errorCondition) { + errorContext.failed = true; + errorContext.extendedStatus = extendedStatus; + LOGS(_log, LOG_LVL_ERROR, context() << "execute" << errorMsg); + } + return errorContext; +} + +void WorkerHttpRequest::init() { + LOGS(_log, LOG_LVL_TRACE, context(__func__)); + replica::Lock lock(_mtx, context(__func__)); + if (status() != protocol::Status::CREATED) return; + + // Start the expiration timer + if (_expirationTimeoutSec != 0) { + _expirationTimer.cancel(); + _expirationTimer.expires_from_now(boost::posix_time::seconds(_expirationTimeoutSec)); + _expirationTimer.async_wait(bind(&WorkerHttpRequest::_expired, shared_from_this(), _1)); + + LOGS(_log, LOG_LVL_TRACE, + context() << __func__ << " started expiration timer with " + << " _expirationTimeoutSec: " << _expirationTimeoutSec); + } +} + +void WorkerHttpRequest::start() { + LOGS(_log, LOG_LVL_TRACE, context(__func__)); + replica::Lock lock(_mtx, context(__func__)); + switch (status()) { + case protocol::Status::CREATED: + setStatus(lock, protocol::Status::IN_PROGRESS); + break; + default: + throw logic_error(context(__func__) + + " not allowed while in status: " + protocol::toString(status())); + } +} + +void WorkerHttpRequest::cancel() { + LOGS(_log, LOG_LVL_TRACE, context(__func__)); + replica::Lock lock(_mtx, context(__func__)); + switch (status()) { + case protocol::Status::QUEUED: + case protocol::Status::CREATED: + case protocol::Status::CANCELLED: + setStatus(lock, protocol::Status::CANCELLED); + break; + case protocol::Status::IN_PROGRESS: + case protocol::Status::IS_CANCELLING: + setStatus(lock, protocol::Status::IS_CANCELLING); + break; + + // Nothing to be done to the completed requests + case protocol::Status::SUCCESS: + case protocol::Status::BAD: + case protocol::Status::FAILED: + break; + } +} + +void WorkerHttpRequest::rollback() { + LOGS(_log, LOG_LVL_TRACE, context(__func__)); + replica::Lock lock(_mtx, context(__func__)); + switch (status()) { + case protocol::Status::CREATED: + case protocol::Status::IN_PROGRESS: + setStatus(lock, protocol::Status::CREATED); + break; + case protocol::Status::IS_CANCELLING: + setStatus(lock, protocol::Status::CANCELLED); + throw WorkerHttpRequestCancelled(); + break; + default: + throw logic_error(context(__func__) + + " not allowed while in status: " + protocol::toString(status())); + } +} + +void WorkerHttpRequest::stop() { + LOGS(_log, LOG_LVL_TRACE, context(__func__)); + replica::Lock lock(_mtx, context(__func__)); + setStatus(lock, protocol::Status::CREATED); +} + +void WorkerHttpRequest::dispose() noexcept { + LOGS(_log, LOG_LVL_TRACE, context(__func__)); + replica::Lock lock(_mtx, context(__func__)); + if (_expirationTimeoutSec != 0) { + try { + _expirationTimer.cancel(); + } catch (exception const& ex) { + LOGS(_log, LOG_LVL_WARN, + context(__func__) << " request expiration couldn't be cancelled, ex: " << ex.what()); + } + } +} + +json WorkerHttpRequest::toJson(bool includeResultIfFinished) const { + LOGS(_log, LOG_LVL_TRACE, context(__func__)); + // IMPORTANT: the lock is not needed here because the data read by the method + // are safe to read w/o any synchronization. The only exception is the results + // which is not a problem since results are only read after the request is finished. + json response; + response["hdr"] = _hdr.toJson(); + response["req"] = _req; + response["type"] = _type; + response["status"] = protocol::toString(_status); + response["status_ext"] = protocol::toString(_extendedStatus); + response["expiration_timeout_sec"] = _expirationTimeoutSec; + response["performance"] = _performance.toJson(); + response["result"] = json::object(); + if (includeResultIfFinished && _status == protocol::Status::SUCCESS) { + getResult(response["result"]); + } + return response; +} + +void WorkerHttpRequest::setStatus(replica::Lock const& lock, protocol::Status status, + protocol::StatusExt extendedStatus) { + LOGS(_log, LOG_LVL_TRACE, + context(__func__) << " " << protocol::toString(_status, _extendedStatus) << " -> " + << protocol::toString(status, extendedStatus)); + switch (status) { + case protocol::Status::CREATED: + _performance.start_time = 0; + _performance.finish_time = 0; + break; + case protocol::Status::IN_PROGRESS: + _performance.setUpdateStart(); + _performance.finish_time = 0; + break; + case protocol::Status::IS_CANCELLING: + break; + case protocol::Status::CANCELLED: + + // Set the start time to some meaningful value in case if the request was + // cancelled while sitting in the input queue + if (0 == _performance.start_time) _performance.setUpdateStart(); + _performance.setUpdateFinish(); + break; + + case protocol::Status::SUCCESS: + case protocol::Status::FAILED: + _performance.setUpdateFinish(); + break; + default: + throw logic_error(context(__func__) + " unhandled status: " + protocol::toString(status)); + } + + // ATTENTION: the top-level status is the last to be modified in + // the state transition to ensure clients will see a consistent state + // of the object. + _extendedStatus = extendedStatus; + _status = status; +} + +void WorkerHttpRequest::_expired(boost::system::error_code const& ec) { + LOGS(_log, LOG_LVL_TRACE, + context() << __func__ << (ec == boost::asio::error::operation_aborted ? " ** ABORTED **" : "")); + replica::Lock lock(_mtx, context(__func__)); + + // Clearing the stored callback after finishing the up-stream notification + // has two purposes: + // + // 1. it guaranties no more than one time notification + // 2. it breaks the up-stream dependency on a caller object if a shared + // pointer to the object was mentioned as the lambda-function's closure + + // Ignore this event if the timer was aborted + if (ec != boost::asio::error::operation_aborted) { + if (_onExpired != nullptr) { + serviceProvider()->io_service().post(bind(move(_onExpired), _hdr.id)); + } + } + _onExpired = nullptr; +} + +} // namespace lsst::qserv::replica diff --git a/src/replica/worker/WorkerHttpRequest.h b/src/replica/worker/WorkerHttpRequest.h new file mode 100644 index 000000000..aeda2d231 --- /dev/null +++ b/src/replica/worker/WorkerHttpRequest.h @@ -0,0 +1,355 @@ +/* + * LSST Data Management System + * + * This product includes software developed by the + * LSST Project (http://www.lsst.org/). + * + * This program is free software: you can redistribute it and/or modify + * it under the terms of the GNU General Public License as published by + * the Free Software Foundation, either version 3 of the License, or + * (at your option) any later version. + * + * This program is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * GNU General Public License for more details. + * + * You should have received a copy of the LSST License Statement and + * the GNU General Public License along with this program. If not, + * see . + */ +#ifndef LSST_QSERV_REPLICA_WORKERHTTPREQUEST_H +#define LSST_QSERV_REPLICA_WORKERHTTPREQUEST_H + +// System headers +#include +#include +#include +#include +#include + +// Third party headers +#include "boost/asio.hpp" + +// Qserv headers +#include "replica/proto/Protocol.h" +#include "replica/util/Common.h" +#include "replica/util/Mutex.h" +#include "replica/util/Performance.h" + +// Forward declarations +namespace lsst::qserv::replica { +class ServiceProvider; +} // namespace lsst::qserv::replica + +// This header declarations +namespace lsst::qserv::replica { + +/** + * Structure WorkerHttpRequestCancelled represent an exception thrown when + * a replication request is cancelled + */ +class WorkerHttpRequestCancelled : public std::exception { +public: + /// @return a short description of the exception + char const* what() const noexcept override { return "cancelled"; } +}; + +/** + * Class WorkerHttpRequest is the base class for a family of the worker-side + * requests which require non-deterministic interactions with the server's + * environment (network, disk I/O, etc.). Generally speaking, all requests + * which can't be implemented instantaneously fall into this category. + */ +class WorkerHttpRequest : public std::enable_shared_from_this { +public: + /// The function type for notifications on the expiration of the request + /// given its unique identifier. + typedef std::function ExpirationCallbackType; + + WorkerHttpRequest() = delete; + WorkerHttpRequest(WorkerHttpRequest const&) = delete; + WorkerHttpRequest& operator=(WorkerHttpRequest const&) = delete; + + /// Destructor (can't 'override' because the base class's one is not virtual) + /// Also, non-trivial destructor is needed to stop the request expiration + /// timer (if any was started by the constructor). + virtual ~WorkerHttpRequest(); + + std::shared_ptr const& serviceProvider() const { return _serviceProvider; } + std::string const& worker() const { return _worker; } + std::string const& type() const { return _type; } + std::string const& id() const { return _hdr.id; } + int priority() const { return _hdr.priority; } + nlohmann::json const& req() const { return _req; } + protocol::Status status() const { return _status; } + protocol::StatusExt extendedStatus() const { return _extendedStatus; } + + WorkerPerformance const& performance() const { return _performance; } + + /** + * This method is called from the initial state protocol::Status::CREATED in order + * to start the request expiration timer. It's safe to call this operation + * multiple times. Each invocation of the method will result in cancelling + * the previously set timer (if any) and starting a new one. + */ + void init(); + + /** + * This method is called from the initial state protocol::Status::CREATED in order + * to prepare the request for processing (to respond to methods 'execute', + * 'cancel', 'rollback' or 'reset'. The final state upon the completion + * of the method should be protocol::Status::IN_PROGRESS. + */ + void start(); + + /** + * This method should be invoked (repeatedly) to execute the request until + * it returns 'true' or throws an exception. Note that returning 'true' + * may mean both success or failure, depending on the completion status + * of the request. + * + * This method is required to be called while the request state is protocol::Status::IN_PROGRESS. + * The method will throw custom exception WorkerHttpRequestCancelled when it detects a cancellation + * request. + * + * @return result of the operation as explained above + */ + virtual bool execute() = 0; + + /** + * Cancel execution of the request. + * + * The effect of the operation varies depending on the current state of + * the request. The default (the base class's implementation) assumes + * the following transitions: + * + * {protocol::Status::CREATED,protocol::Status::CANCELLED} -> protocol::Status::CANCELLED + * {protocol::Status::IN_PROGRESS,protocol::Status::IS_CANCELLING} -> protocol::Status::IS_CANCELLING + * {*} -> throw std::logic_error + */ + virtual void cancel(); + + /** + * Roll back the request into its initial state and cleanup partial results + * if possible. + * + * The effect of the operation varies depending on the current state of + * the request. The default (the base class's implementation) assumes + * the following transitions: + * + * {protocol::Status::CREATED, protocol::Status::IN_PROGRESS} -> protocol::Status::CREATED + * {protocol::Status::IS_CANCELLING} -> protocol::Status::CANCELLED -> throw WorkerHttpRequestCancelled + * {*} -> throw std::logic_error + */ + virtual void rollback(); + + /** + * This method is called from *ANY* initial state in order to turn + * the request back into the initial protocol::Status::CREATED. + * @param func (optional) the name of a function/method which requested the context string + */ + void stop(); + + /** + * This method should be used to cancel the request expiration timer. + * Normally this method is initiated during the external "garbage collection" + * of requests to ensure all resources (including a copy of a smart pointer onto + * objects of the request classes) held by timers get released. + * + * @note this method won't throw any exceptions so that it could + * be invoked from the destructor. All exceptions (should they + * occur during an execution of the method) will be intersected + * and reported as errors to the message logger. + */ + void dispose() noexcept; + + /** + * Extract the extra data from the request and put it into the response object. + * @param includeResultIfFinished (optional) flag to include results if the request has finished + */ + nlohmann::json toJson(bool includeResultIfFinished = false) const; + + /// @return the context string + std::string context(std::string const& func = std::string()) const { + return id() + " " + type() + " " + protocol::toString(status()) + " " + func; + } + +protected: + /** + * The normal constructor of the class + * + * @param serviceProvider provider is needed to access the Configuration of + * a setup and for validating the input parameters + * @param worker the name of a worker. It must be the same worker as the one + * where the request is going to be processed. + * @param type the type name of a request + * @param hdr request header (common parameters of the queued request) + * @param req the request object received from a client (request-specific parameters) + * @param onExpired request expiration callback function + * @throws std::invalid_argument if the worker is unknown + */ + WorkerHttpRequest(std::shared_ptr const& serviceProvider, std::string const& worker, + std::string const& type, protocol::QueuedRequestHdr const& hdr, + nlohmann::json const& req, ExpirationCallbackType const& onExpired); + + /** + * The method is used to check if the request is entered the cancellation state. + * The implementation assumes the following transitions: + * + * {protocol::Status::IN_PROGRESS} -> protocol::Status::IN_PROGRESS + * {protocol::Status::IS_CANCELLING} -> protocol::Status::CANCELLED -> throw WorkerHttpRequestCancelled + * {*} -> throw std::logic_error + * + * @param lock a lock on _mtx which acquired before calling this method + * @param func the name of a function/method which called the method + * @throws WorkerHttpRequestCancelled if the request is being cancelled. + * @throws std::logic_error if the state is not as expected. + */ + void checkIfCancelling(replica::Lock const& lock, std::string const& func); + + /** Set the status + * + * @note this method needs to be called within a thread-safe context + * when moving requests between different queues. + * + * @param lock a lock which acquired before calling this method + * @param status primary status to be set + * @param extendedStatus secondary status to be set + */ + void setStatus(replica::Lock const& lock, protocol::Status status, + protocol::StatusExt extendedStatus = protocol::StatusExt::NONE); + + /** + * Fill in the information object for the specified request based on its + * actual type. + * @param result an object to be filled + */ + virtual void getResult(nlohmann::json& result) const = 0; + + /** + * Structure ErrorContext is used for tracking errors reported by + * method 'reportErrorIf + */ + struct ErrorContext { + // State of the object + bool failed; + protocol::StatusExt extendedStatus; + + ErrorContext() : failed(false), extendedStatus(protocol::StatusExt::NONE) {} + + /** + * Merge the context of another object into the current one. + * + * @note Only the first error code will be stored when a error condition + * is detected. An assumption is that the first error would usually cause + * a "chain reaction", hence only the first one typically matters. + * Other details could be found in the log files if needed. + * @param ErrorContext input context to be merged with the current state + */ + ErrorContext& operator||(const ErrorContext& rhs) { + if (&rhs != this) { + if (rhs.failed and not failed) { + failed = true; + extendedStatus = rhs.extendedStatus; + } + } + return *this; + } + }; + + /** + * Check if the error condition is set and report the error. + * The error message will be sent to the corresponding logging + * stream. + * + * @param condition if set to 'true' then there is a error condition + * @param extendedStatus extended status corresponding to the condition + * (will be ignored if no error condition is present) + * @param errorMsg a message to be reported into the log stream + * @return the context object encapsulating values passed in parameters + * 'condition' and 'extendedStatus' + */ + ErrorContext reportErrorIf(bool condition, protocol::StatusExt extendedStatus, + std::string const& errorMsg); + + /// Return shared pointer of the desired subclass (no dynamic type checking) + template + std::shared_ptr shared_from_base() { + return std::static_pointer_cast(shared_from_this()); + } + + // Input parameters + + std::shared_ptr const _serviceProvider; + + std::string const _worker; + std::string const _type; + protocol::QueuedRequestHdr const _hdr; + nlohmann::json const _req; + + ExpirationCallbackType _onExpired; ///< The callback is reset when the request gets expired + /// or explicitly disposed. + unsigned int const _expirationTimeoutSec; + + /// This timer is used (if configured) to limit the total duration of time + /// a request could exist from its creation till termination. The timer + /// starts when the request gets created. And it's explicitly finished when + /// a request object gets destroyed. + /// + /// If the time has a chance to expire then the request expiration callback + /// (if any) passed into the constructor will be invoked to notify WorkerProcessor + /// on the expiration event. + boost::asio::deadline_timer _expirationTimer; + + // 2-layer state of a request + + std::atomic _status; + std::atomic _extendedStatus; + + /// Performance counters + WorkerPerformance _performance; + + /// Mutex guarding API calls where it's needed + mutable replica::Mutex _mtx; + + /// Mutex guarding operations with the worker's data folder + static replica::Mutex _mtxDataFolderOperations; + +private: + /** + * Request expiration timer's handler. The expiration interval (if any) + * is obtained from the Controller-side requests or obtained from + * the configuration service. When the request expires (and if the timer + * is not aborted due to request disposal) then an upstream callback + * is invoked. + * + * @param ec error code to be checked to see if the time was aborted + * by the explicit request disposal operation. + */ + void _expired(boost::system::error_code const& ec); + + // For memory usage monitoring and memory leak diagnostic. + static std::atomic _numInstances; +}; + +/** + * Structure WorkerHttpRequestCompare is a functor representing a comparison type + * for strict weak ordering required by std::priority_queue + */ +struct WorkerHttpRequestCompare { + /** + * Sort requests by their priorities + * @param lhs pointer to a request on the left side of a logical comparison + * @param rhs pointer to a request on the right side of a logical comparison + * @return 'true' if the priority of 'lhs' is strictly less than the one of 'rhs' + */ + bool operator()(std::shared_ptr const& lhs, + std::shared_ptr const& rhs) const { + return lhs->priority() < rhs->priority(); + } +}; + +} // namespace lsst::qserv::replica + +#endif // LSST_QSERV_REPLICA_WORKERHTTPREQUEST_H diff --git a/src/replica/worker/WorkerHttpSvc.cc b/src/replica/worker/WorkerHttpSvc.cc new file mode 100644 index 000000000..d7151cc2f --- /dev/null +++ b/src/replica/worker/WorkerHttpSvc.cc @@ -0,0 +1,146 @@ +/* + * LSST Data Management System + * + * This product includes software developed by the + * LSST Project (http://www.lsst.org/). + * + * This program is free software: you can redistribute it and/or modify + * it under the terms of the GNU General Public License as published by + * the Free Software Foundation, either version 3 of the License, or + * (at your option) any later version. + * + * This program is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * GNU General Public License for more details. + * + * You should have received a copy of the LSST License Statement and + * the GNU General Public License along with this program. If not, + * see . + */ + +// Class header +#include "replica/worker/WorkerHttpSvc.h" + +// System headers +#include +#include + +// Qserv headers +#include "http/ChttpMetaModule.h" +#include "replica/config/Configuration.h" +#include "replica/services/ServiceProvider.h" +#include "replica/util/Common.h" +#include "replica/worker/WorkerHttpProcessor.h" +#include "replica/worker/WorkerHttpSvcMod.h" + +// LSST headers +#include "lsst/log/Log.h" + +// Third party headers +#include "httplib.h" +#include "nlohmann/json.hpp" + +using namespace nlohmann; +using namespace std; + +namespace { +string const context_ = "WORKER-HTTP-SVC "; +LOG_LOGGER _log = LOG_GET("lsst.qserv.worker.WorkerHttpSvc"); +} // namespace + +namespace lsst::qserv::replica { + +shared_ptr WorkerHttpSvc::create(shared_ptr const& serviceProvider, + string const& workerName) { + return shared_ptr(new WorkerHttpSvc(serviceProvider, workerName)); +} + +WorkerHttpSvc::WorkerHttpSvc(shared_ptr const& serviceProvider, string const& workerName) + : ChttpSvc(context_, serviceProvider, + serviceProvider->config()->get("worker", "http-svc-port"), + serviceProvider->config()->get("worker", "http-svc-max-queued-requests"), + serviceProvider->config()->get("worker", "num-http-svc-threads")), + _workerName(workerName), + _processor(WorkerHttpProcessor::create(serviceProvider, workerName)) {} + +void WorkerHttpSvc::registerServices(unique_ptr const& server) { + throwIf(server == nullptr, context_ + "the server is not initialized"); + auto const self = shared_from_base(); + server->Get("/meta/version", [self](httplib::Request const& req, httplib::Response& resp) { + json const info = json::object({{"kind", "replication-worker-svc"}, + {"id", self->_workerName}, + {"instance_id", self->serviceProvider()->instanceId()}}); + http::ChttpMetaModule::process(context_, info, req, resp, "VERSION"); + }); + server->Post("/worker/echo", [self](httplib::Request const& req, httplib::Response& resp) { + WorkerHttpSvcMod::process(self->serviceProvider(), self->_processor, self->_workerName, req, resp, + "ECHO", http::AuthType::REQUIRED); + }); + server->Post("/worker/replica/create", [self](httplib::Request const& req, httplib::Response& resp) { + WorkerHttpSvcMod::process(self->serviceProvider(), self->_processor, self->_workerName, req, resp, + "REPLICA-CREATE", http::AuthType::REQUIRED); + }); + server->Post("/worker/replica/delete", [self](httplib::Request const& req, httplib::Response& resp) { + WorkerHttpSvcMod::process(self->serviceProvider(), self->_processor, self->_workerName, req, resp, + "REPLICA-DELETE", http::AuthType::REQUIRED); + }); + server->Post("/worker/replica/find", [self](httplib::Request const& req, httplib::Response& resp) { + WorkerHttpSvcMod::process(self->serviceProvider(), self->_processor, self->_workerName, req, resp, + "REPLICA-FIND", http::AuthType::REQUIRED); + }); + server->Post("/worker/replica/find-all", [self](httplib::Request const& req, httplib::Response& resp) { + WorkerHttpSvcMod::process(self->serviceProvider(), self->_processor, self->_workerName, req, resp, + "REPLICA-FIND-ALL", http::AuthType::REQUIRED); + }); + server->Post("/worker/index", [self](httplib::Request const& req, httplib::Response& resp) { + WorkerHttpSvcMod::process(self->serviceProvider(), self->_processor, self->_workerName, req, resp, + "INDEX", http::AuthType::REQUIRED); + }); + server->Post("/worker/sql", [self](httplib::Request const& req, httplib::Response& resp) { + WorkerHttpSvcMod::process(self->serviceProvider(), self->_processor, self->_workerName, req, resp, + "SQL", http::AuthType::REQUIRED); + }); + server->Get("/worker/request/track/:id", [self](httplib::Request const& req, httplib::Response& resp) { + WorkerHttpSvcMod::process(self->serviceProvider(), self->_processor, self->_workerName, req, resp, + "REQUEST-TRACK"); + }); + server->Get("/worker/request/status/:id", [self](httplib::Request const& req, httplib::Response& resp) { + WorkerHttpSvcMod::process(self->serviceProvider(), self->_processor, self->_workerName, req, resp, + "REQUEST-STATUS"); + }); + server->Put("/worker/request/stop/:id", [self](httplib::Request const& req, httplib::Response& resp) { + WorkerHttpSvcMod::process(self->serviceProvider(), self->_processor, self->_workerName, req, resp, + "REQUEST-STOP", http::AuthType::REQUIRED); + }); + server->Delete("/worker/request/dispose", [self](httplib::Request const& req, httplib::Response& resp) { + WorkerHttpSvcMod::process(self->serviceProvider(), self->_processor, self->_workerName, req, resp, + "REQUEST-DISPOSE", http::AuthType::REQUIRED); + }); + server->Get("/worker/service", [self](httplib::Request const& req, httplib::Response& resp) { + WorkerHttpSvcMod::process(self->serviceProvider(), self->_processor, self->_workerName, req, resp, + "SERVICE-STATUS"); + }); + server->Get("/worker/service/requests", [self](httplib::Request const& req, httplib::Response& resp) { + WorkerHttpSvcMod::process(self->serviceProvider(), self->_processor, self->_workerName, req, resp, + "SERVICE-REQUESTS"); + }); + server->Put("/worker/service/suspend", [self](httplib::Request const& req, httplib::Response& resp) { + WorkerHttpSvcMod::process(self->serviceProvider(), self->_processor, self->_workerName, req, resp, + "SERVICE-SUSPEND", http::AuthType::REQUIRED); + }); + server->Put("/worker/service/resume", [self](httplib::Request const& req, httplib::Response& resp) { + WorkerHttpSvcMod::process(self->serviceProvider(), self->_processor, self->_workerName, req, resp, + "SERVICE-RESUME", http::AuthType::REQUIRED); + }); + server->Put("/worker/service/drain", [self](httplib::Request const& req, httplib::Response& resp) { + WorkerHttpSvcMod::process(self->serviceProvider(), self->_processor, self->_workerName, req, resp, + "SERVICE-DRAIN", http::AuthType::REQUIRED); + }); + server->Put("/worker/service/reconfig", [self](httplib::Request const& req, httplib::Response& resp) { + WorkerHttpSvcMod::process(self->serviceProvider(), self->_processor, self->_workerName, req, resp, + "SERVICE-RECONFIG", http::AuthType::REQUIRED); + }); +} + +} // namespace lsst::qserv::replica diff --git a/src/replica/worker/WorkerHttpSvc.h b/src/replica/worker/WorkerHttpSvc.h new file mode 100644 index 000000000..0e204649e --- /dev/null +++ b/src/replica/worker/WorkerHttpSvc.h @@ -0,0 +1,84 @@ +/* + * LSST Data Management System + * + * This product includes software developed by the + * LSST Project (http://www.lsst.org/). + * + * This program is free software: you can redistribute it and/or modify + * it under the terms of the GNU General Public License as published by + * the Free Software Foundation, either version 3 of the License, or + * (at your option) any later version. + * + * This program is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * GNU General Public License for more details. + * + * You should have received a copy of the LSST License Statement and + * the GNU General Public License along with this program. If not, + * see . + */ +#ifndef LSST_QSERV_REPLICA_WORKERHTTPSVC_H +#define LSST_QSERV_REPLICA_WORKERHTTPSVC_H + +// System headers +#include +#include + +// Qserv headers +#include "replica/util/ChttpSvc.h" + +// Forward declarations +namespace lsst::qserv::replica { +class ServiceProvider; +class WorkerHttpProcessor; +} // namespace lsst::qserv::replica + +namespace httplib { +class Server; +} // namespace httplib + +// This header declarations +namespace lsst::qserv::replica { + +/** + * Class WorkerHttpSvc is the HTTP frontend to the Replication Worker Service. + * Each instance of this class will be running in its own thread. + */ +class WorkerHttpSvc : public ChttpSvc { +public: + /** + * Create an instance of the service. + * + * @param serviceProvider For configuration, etc. services. + * @param workerName The name of a worker this service is acting upon (used for + * checking consistency of the protocol). + * @return A pointer to the created object. + */ + static std::shared_ptr create(std::shared_ptr const& serviceProvider, + std::string const& workerName); + + WorkerHttpSvc() = delete; + WorkerHttpSvc(WorkerHttpSvc const&) = delete; + WorkerHttpSvc& operator=(WorkerHttpSvc const&) = delete; + + virtual ~WorkerHttpSvc() = default; + +protected: + /// @see HttpSvc::registerServices() + virtual void registerServices(std::unique_ptr const& server) override; + +private: + /// @see WorkerHttpSvc::create() + WorkerHttpSvc(std::shared_ptr const& serviceProvider, std::string const& workerName); + + // Input parameters + std::string const _workerName; + + /// The request processor. + std::shared_ptr _processor; +}; + +} // namespace lsst::qserv::replica + +#endif // LSST_QSERV_REPLICA_WORKERHTTPSVC_H diff --git a/src/replica/worker/WorkerHttpSvcMod.cc b/src/replica/worker/WorkerHttpSvcMod.cc new file mode 100644 index 000000000..37812e177 --- /dev/null +++ b/src/replica/worker/WorkerHttpSvcMod.cc @@ -0,0 +1,245 @@ +/* + * LSST Data Management System + * + * This product includes software developed by the + * LSST Project (http://www.lsst.org/). + * + * This program is free software: you can redistribute it and/or modify + * it under the terms of the GNU General Public License as published by + * the Free Software Foundation, either version 3 of the License, or + * (at your option) any later version. + * + * This program is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * GNU General Public License for more details. + * + * You should have received a copy of the LSST License Statement and + * the GNU General Public License along with this program. If not, + * see . + */ + +// Class header +#include "replica/worker/WorkerHttpSvcMod.h" + +// System headers +#include + +// Third-party headers +#include + +// Qserv header +#include "http/Method.h" +#include "replica/proto/Protocol.h" +#include "replica/worker/WorkerHttpProcessor.h" +#include "replica/services/ServiceProvider.h" + +using namespace std; +using json = nlohmann::json; + +namespace lsst::qserv::replica { + +void WorkerHttpSvcMod::process(shared_ptr const& serviceProvider, + shared_ptr const& processor, string const& workerName, + httplib::Request const& req, httplib::Response& resp, + string const& subModuleName, http::AuthType const authType) { + WorkerHttpSvcMod module(serviceProvider, processor, workerName, req, resp); + module.execute(subModuleName, authType); +} + +WorkerHttpSvcMod::WorkerHttpSvcMod(shared_ptr const& serviceProvider, + shared_ptr const& processor, string const& workerName, + httplib::Request const& req, httplib::Response& resp) + : http::ChttpModule(serviceProvider->authKey(), serviceProvider->adminAuthKey(), req, resp), + _serviceProvider(serviceProvider), + _processor(processor), + _workerName(workerName) {} + +string WorkerHttpSvcMod::context() const { return "WORKER-HTTP-SVC "; } + +json WorkerHttpSvcMod::executeImpl(string const& subModuleName) { + debug(__func__, "subModuleName: '" + subModuleName + "'"); + enforceInstanceId(__func__, _serviceProvider->instanceId()); + if (subModuleName == "ECHO") + return _echo(); + else if (subModuleName == "REPLICA-CREATE") + return _replicaCreate(); + else if (subModuleName == "REPLICA-DELETE") + return _replicaDelete(); + else if (subModuleName == "REPLICA-FIND") + return _replicaFind(); + else if (subModuleName == "REPLICA-FIND-ALL") + return _replicaFindAll(); + else if (subModuleName == "SQL") + return _sql(); + else if (subModuleName == "INDEX") + return _index(); + else if (subModuleName == "REQUEST-TRACK") + return _requestTrack(); + else if (subModuleName == "REQUEST-STATUS") + return _requestStatus(); + else if (subModuleName == "REQUEST-STOP") + return _requestStop(); + else if (subModuleName == "REQUEST-DISPOSE") + return _requestDispose(); + else if (subModuleName == "SERVICE-SUSPEND") + return _serviceSuspend(); + else if (subModuleName == "SERVICE-RESUME") + return _serviceResume(); + else if (subModuleName == "SERVICE-STATUS") + return _serviceStatus(); + else if (subModuleName == "SERVICE-REQUESTS") + return _serviceRequests(); + else if (subModuleName == "SERVICE-DRAIN") + return _serviceDrain(); + else if (subModuleName == "SERVICE-RECONFIG") + return _serviceReconfig(); + throw invalid_argument(context() + "::" + string(__func__) + " unsupported sub-module: '" + + subModuleName + "'"); +} + +protocol::QueuedRequestHdr WorkerHttpSvcMod::_parseHdr(string const& func) const { + auto const hdrJson = body().required("hdr"); + protocol::QueuedRequestHdr const hdr(body().required(hdrJson, "id"), + body().optional(hdrJson, "priority", 0), + body().optional(hdrJson, "timeout", 0)); + debug(func, "id: '" + hdr.id + "'"); + debug(func, "priority: " + to_string(hdr.priority)); + debug(func, "timeout: " + to_string(hdr.timeout)); + return hdr; +} + +json WorkerHttpSvcMod::_echo() const { + debug(__func__); + checkApiVersion(__func__, 40); + return _processor->echo(_parseHdr(__func__), body().required("req")); +} + +json WorkerHttpSvcMod::_replicaCreate() { + debug(__func__); + checkApiVersion(__func__, 40); + return _processor->createReplica(_parseHdr(__func__), body().required("req")); +} + +json WorkerHttpSvcMod::_replicaDelete() { + debug(__func__); + checkApiVersion(__func__, 40); + return _processor->deleteReplica(_parseHdr(__func__), body().required("req")); +} + +json WorkerHttpSvcMod::_replicaFind() { + debug(__func__); + checkApiVersion(__func__, 40); + return _processor->findReplica(_parseHdr(__func__), body().required("req")); +} + +json WorkerHttpSvcMod::_replicaFindAll() { + debug(__func__); + checkApiVersion(__func__, 40); + return _processor->findAllReplicas(_parseHdr(__func__), body().required("req")); +} + +json WorkerHttpSvcMod::_index() { + debug(__func__); + checkApiVersion(__func__, 40); + return _processor->index(_parseHdr(__func__), body().required("req")); +} + +json WorkerHttpSvcMod::_sql() { + debug(__func__); + checkApiVersion(__func__, 40); + return _processor->sql(_parseHdr(__func__), body().required("req")); +} + +json WorkerHttpSvcMod::_requestTrack() { + debug(__func__); + checkApiVersion(__func__, 40); + string const id = params().at("id"); + debug(__func__, "id: '" + id + "'"); + return _processor->trackRequest(id); +} + +json WorkerHttpSvcMod::_requestStatus() { + debug(__func__); + checkApiVersion(__func__, 40); + string const id = params().at("id"); + debug(__func__, "id: '" + id + "'"); + return _processor->requestStatus(id); +} + +json WorkerHttpSvcMod::_requestStop() { + debug(__func__); + checkApiVersion(__func__, 40); + string const id = params().at("id"); + debug(__func__, "id: '" + id + "'"); + return _processor->stopRequest(id); +} + +json WorkerHttpSvcMod::_requestDispose() { + debug(__func__); + checkApiVersion(__func__, 40); + auto const idsJson = body().required("ids"); + if (!idsJson.is_array()) + throw invalid_argument(context() + "::" + string(__func__) + " 'ids' is not an array"); + + json idsDisposedJson = json::object(); + for (auto const& idJson : idsJson) { + string const id = idJson.get(); + idsDisposedJson[id] = _processor->disposeRequest(id) ? 1 : 0; + } + return json::object({{"status", protocol::Status::SUCCESS}, + {"status_ext", protocol::StatusExt::NONE}, + {"ids_disposed", idsDisposedJson}}); +} + +json WorkerHttpSvcMod::_serviceSuspend() { + debug(__func__); + checkApiVersion(__func__, 40); + + // This operation is allowed to be asynchronous as it may take + // extra time for the processor's threads to finish on-going processing + _processor->stop(); + return _processor->toJson(_processor->state() == protocol::ServiceState::RUNNING + ? protocol::Status::FAILED + : protocol::Status::SUCCESS); +} + +json WorkerHttpSvcMod::_serviceResume() { + debug(__func__); + checkApiVersion(__func__, 40); + _processor->run(); + return _processor->toJson(_processor->state() == protocol::ServiceState::RUNNING + ? protocol::Status::SUCCESS + : protocol::Status::FAILED); +} + +json WorkerHttpSvcMod::_serviceStatus() { + debug(__func__); + checkApiVersion(__func__, 40); + return _processor->toJson(protocol::Status::SUCCESS); +} + +json WorkerHttpSvcMod::_serviceRequests() { + debug(__func__); + checkApiVersion(__func__, 40); + const bool includeRequests = true; + return _processor->toJson(protocol::Status::SUCCESS, includeRequests); +} + +json WorkerHttpSvcMod::_serviceDrain() { + debug(__func__); + checkApiVersion(__func__, 40); + _processor->drain(); + const bool includeRequests = true; + return _processor->toJson(protocol::Status::SUCCESS, includeRequests); +} + +json WorkerHttpSvcMod::_serviceReconfig() { + debug(__func__); + checkApiVersion(__func__, 40); + _processor->reconfig(); + const bool includeRequests = true; + return _processor->toJson(protocol::Status::SUCCESS, includeRequests); +} + +} // namespace lsst::qserv::replica diff --git a/src/replica/worker/WorkerHttpSvcMod.h b/src/replica/worker/WorkerHttpSvcMod.h new file mode 100644 index 000000000..bf72ad0bd --- /dev/null +++ b/src/replica/worker/WorkerHttpSvcMod.h @@ -0,0 +1,172 @@ +/* + * LSST Data Management System + * + * This product includes software developed by the + * LSST Project (http://www.lsst.org/). + * + * This program is free software: you can redistribute it and/or modify + * it under the terms of the GNU General Public License as published by + * the Free Software Foundation, either version 3 of the License, or + * (at your option) any later version. + * + * This program is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * GNU General Public License for more details. + * + * You should have received a copy of the LSST License Statement and + * the GNU General Public License along with this program. If not, + * see . + */ +#ifndef LSST_QSERV_WORKERHTTPSVCMOD_H +#define LSST_QSERV_WORKERHTTPSVCMOD_H + +// System headers +#include + +// Third party headers +#include "nlohmann/json.hpp" + +// Qserv headers +#include "http/ChttpModule.h" + +// Forward declarations + +namespace lsst::qserv::replica { +class ServiceProvider; +class WorkerHttpProcessor; +} // namespace lsst::qserv::replica + +namespace lsst::qserv::replica::protocol { +struct QueuedRequestHdr; +} // namespace lsst::qserv::replica::protocol + +// This header declarations +namespace lsst::qserv::replica { + +/** + * Class WorkerHttpSvcMod processes the Replication Controller's requests. + * The class is used by the HTTP server built into the worker Replication service. + */ +class WorkerHttpSvcMod : public http::ChttpModule { +public: + WorkerHttpSvcMod() = delete; + WorkerHttpSvcMod(WorkerHttpSvcMod const&) = delete; + WorkerHttpSvcMod& operator=(WorkerHttpSvcMod const&) = delete; + + virtual ~WorkerHttpSvcMod() = default; + + /** + * Process a request. + * + * Supported values for parameter 'subModuleName': + * + * ECHO for testing the worker-side framework + * REPLICA-CREATE for creating a replica of a chunk + * REPLICA-DELETE for deleting an existing replica of a chunk + * REPLICA-FIND for finding out if a replica is present, and reporting its state + * REPLICA-FIND-ALL for finding all replicas and reporting their states + * INDEX for extracting and returning a collection of the "director" index data + * SQL for executing various SQL statements against the worker's database + * REQUEST-TRACK for tracking status and retreiving results of the previously submitted request + * REQUEST-STATUS for checking the status of the previously submitted request + * REQUEST-STOP for stopping the previously submitted request + * REQUEST-DISPOSE for garbage collecting the request + * SERVICE-STATUS for checking the status of the worker replication service + * SERVICE-SUSPEND for suspending the worker replication service + * SERVICE-RESUME for resuming the worker replication service + * SERVICE-REQUESTS for listing the outstanding requests + * SERVICE-DRAIN for draining the worker replication service + * SERVICE-RECONFIG for reconfiguring the worker replication service + * + * @param serviceProvider The provider of services is needed to access + * the configuration and the database services. + * @param workerName The name of a worker this service is acting upon (used to pull + * worker-specific configuration options for the service). + * @param processor Request processor. + * @param req The HTTP request. + * @param resp The HTTP response channel. + * @param subModuleName The name of a submodule to be called. + * @param authType The authorization requirements for the module + * @throws std::invalid_argument for unknown values of parameter 'subModuleName' + */ + static void process(std::shared_ptr const& serviceProvider, + std::shared_ptr const& processor, std::string const& workerName, + httplib::Request const& req, httplib::Response& resp, + std::string const& subModuleName, + http::AuthType const authType = http::AuthType::NONE); + +protected: + virtual std::string context() const final; + virtual nlohmann::json executeImpl(std::string const& subModuleName) final; + +private: + WorkerHttpSvcMod(std::shared_ptr const& serviceProvider, + std::shared_ptr const& processor, std::string const& workerName, + httplib::Request const& req, httplib::Response& resp); + + /// Parse common parameters of the queued requests + /// @param func The name of the function to be used in the log messages + /// @return The parsed header + protocol::QueuedRequestHdr _parseHdr(std::string const& func) const; + + /// Process the ECHO request + nlohmann::json _echo() const; + + /// Process the REPLICA-CREATE request + nlohmann::json _replicaCreate(); + + /// Process the REPLICA-DELETE request + nlohmann::json _replicaDelete(); + + /// Process the REPLICA-FIND request + nlohmann::json _replicaFind(); + + /// Process the REPLICA-FIND-ALL request + nlohmann::json _replicaFindAll(); + + /// Process the INDEX request + nlohmann::json _index(); + + /// Process the SQL request + nlohmann::json _sql(); + + /// Process the REQUEST-TRACK request + nlohmann::json _requestTrack(); + + /// Process the REQUEST-STATUS request + nlohmann::json _requestStatus(); + + /// Process the REQUEST-STOP request + nlohmann::json _requestStop(); + + /// Process the REQUEST-DISPOSE request + nlohmann::json _requestDispose(); + + /// Process the SERVICE-SUSPEND request + nlohmann::json _serviceSuspend(); + + /// Process the SERVICE-RESUME request + nlohmann::json _serviceResume(); + + /// Process the SERVICE-STATUS request + nlohmann::json _serviceStatus(); + + /// Process the SERVICE-REQUESTS request + nlohmann::json _serviceRequests(); + + /// Process the SERVICE-DRAIN request + nlohmann::json _serviceDrain(); + + /// Process the SERVICE-RECONFIG request + nlohmann::json _serviceReconfig(); + + // Input parameters + std::shared_ptr const _serviceProvider; + std::shared_ptr _processor; + std::string const _workerName; +}; + +} // namespace lsst::qserv::replica + +#endif // LSST_QSERV_WORKERHTTPSVCMOD_H