Extended Replication system's framework (more)

This commit to be squashed with the similar one made earlier.

src/replica/CMakeLists.txt

@@ -76,6 +76,7 @@ target_sources(replica PRIVATE
     GetReplicasQservMgtRequest.cc
     GetDbStatusQservMgtRequest.cc
     GetConfigQservMgtRequest.cc
+    GetStatusQservHttpMgtRequest.cc
     GetStatusQservMgtRequest.cc
     HealthMonitorTask.cc
     HttpAsyncReqApp.cc
@@ -130,6 +131,7 @@ target_sources(replica PRIVATE
     PurgeJob.cc
     QhttpTestApp.cc
     QservGetReplicasJob.cc
+    QservHttpMgtRequest.cc
     QservMgtRequest.cc
     QservMgtServices.cc
     QservStatusJob.cc

src/replica/GetStatusQservHttpMgtRequest.cc

@@ -0,0 +1,123 @@
+/*
+ * 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 <http://www.lsstcorp.org/LegalNotices/>.
+ */
+
+// Class header
+#include "replica/GetStatusQservHttpMgtRequest.h"
+
+// System headers
+#include <set>
+#include <stdexcept>
+
+// Qserv headers
+#include "http/AsyncReq.h"
+
+// LSST headers
+#include "lsst/log/Log.h"
+
+using namespace nlohmann;
+using namespace std;
+
+namespace {
+
+LOG_LOGGER _log = LOG_GET("lsst.qserv.replica.GetStatusQservHttpMgtRequest");
+
+}  // namespace
+
+namespace lsst::qserv::replica {
+
+shared_ptr<GetStatusQservHttpMgtRequest> GetStatusQservHttpMgtRequest::create(
+        shared_ptr<ServiceProvider> const& serviceProvider, string const& worker,
+        wbase::TaskSelector const& taskSelector, GetStatusQservHttpMgtRequest::CallbackType const& onFinish) {
+    return shared_ptr<GetStatusQservHttpMgtRequest>(
+            new GetStatusQservHttpMgtRequest(serviceProvider, worker, taskSelector, onFinish));
+}
+
+GetStatusQservHttpMgtRequest::GetStatusQservHttpMgtRequest(
+        shared_ptr<ServiceProvider> const& serviceProvider, string const& worker,
+        wbase::TaskSelector const& taskSelector, GetStatusQservHttpMgtRequest::CallbackType const& onFinish)
+        : QservHttpMgtRequest(serviceProvider, "QSERV_GET_STATUS", worker),
+          _taskSelector(taskSelector),
+          _onFinish(onFinish) {}
+
+json const& GetStatusQservHttpMgtRequest::info() const {
+    if (!((state() == State::FINISHED) && (extendedState() == ExtendedState::SUCCESS))) {
+        throw logic_error("GetStatusQservHttpMgtRequest::" + string(__func__) +
+                          "  no info available in state: " + state2string(state(), extendedState()));
+    }
+    return _info;
+}
+
+void GetStatusQservHttpMgtRequest::createHttpReqImpl(replica::Lock const& lock) {
+    auto const onHttpReqFinish = [self = shared_from_base<GetStatusQservHttpMgtRequest>()](
+                                         shared_ptr<http::AsyncReq> const& request) {
+        self->_processResponse(request);
+    };
+    string const service = "/status";
+    string const query = wbase::taskSelectorToHttpQuery(_taskSelector);
+    createHttpReq(lock, onHttpReqFinish, service, query);
+}
+
+void GetStatusQservHttpMgtRequest::notify(replica::Lock const& lock) {
+    LOGS(_log, LOG_LVL_TRACE, context() << __func__);
+    notifyDefaultImpl<GetStatusQservHttpMgtRequest>(lock, _onFinish);
+}
+
+void GetStatusQservHttpMgtRequest::_processResponse(shared_ptr<http::AsyncReq> const& request) {
+    string const context_ = context() + string(__func__) + " ";
+    if (state() == State::FINISHED) return;
+    replica::Lock const lock(_mtx, context_);
+    if (state() == State::FINISHED) return;
+
+    switch (request->state()) {
+        case http::AsyncReq::State::FINISHED:
+            try {
+                _info = json::parse(request->responseBody());
+                finish(lock, QservHttpMgtRequest::ExtendedState::SUCCESS);
+            } catch (exception const& ex) {
+                string const msg = "failed to parse worker response, ex: " + string(ex.what());
+                finish(lock, QservHttpMgtRequest::ExtendedState::SERVER_BAD_RESPONSE, msg);
+            }
+            if (_info.at("success").get<int>() == 0) {
+                string const msg = "worker reported error: " + _info.at("error").get<string>();
+                finish(lock, QservHttpMgtRequest::ExtendedState::SERVER_BAD, msg);
+            }
+            break;
+        case http::AsyncReq::State::FAILED:
+            finish(lock, QservHttpMgtRequest::ExtendedState::SERVER_ERROR,
+                   request->errorMessage() + ", code: " + to_string(request->responseCode()));
+            break;
+        case http::AsyncReq::State::BODY_LIMIT_ERROR:
+            finish(lock, QservHttpMgtRequest::ExtendedState::BODY_LIMIT_ERROR,
+                   request->errorMessage() + ", code: " + to_string(request->responseCode()));
+            break;
+        case http::AsyncReq::State::CANCELLED:
+            finish(lock, QservHttpMgtRequest::ExtendedState::CANCELLED);
+            break;
+        case http::AsyncReq::State::EXPIRED:
+            finish(lock, QservHttpMgtRequest::ExtendedState::TIMEOUT_EXPIRED);
+            break;
+        default:
+            throw runtime_error(context_ + "unsupported state of the HTTP request: " +
+                                http::AsyncReq::state2str(request->state()));
+    }
+}
+
+}  // namespace lsst::qserv::replica

src/replica/GetStatusQservHttpMgtRequest.h

@@ -0,0 +1,109 @@
+/*
+ * 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 <http://www.lsstcorp.org/LegalNotices/>.
+ */
+#ifndef LSST_QSERV_REPLICA_GETSTATUSQSERVHTTPMGTREQUEST_H
+#define LSST_QSERV_REPLICA_GETSTATUSQSERVHTTPMGTREQUEST_H
+
+// System headers
+#include <memory>
+#include <string>
+
+// Third party headers
+#include "nlohmann/json.hpp"
+
+// Qserv headers
+#include "replica/QservHttpMgtRequest.h"
+#include "wbase/TaskState.h"
+
+namespace lsst::qserv::replica {
+class ServiceProvider;
+}  // namespace lsst::qserv::replica
+
+// This header declarations
+namespace lsst::qserv::replica {
+
+/**
+ * Class GetStatusQservHttpMgtRequest is a request for obtaining various info
+ * (status, counters, monitoring) reported the Qserv workers.
+ */
+class GetStatusQservHttpMgtRequest : public QservHttpMgtRequest {
+public:
+    /// The function type for notifications on the completion of the request
+    typedef std::function<void(std::shared_ptr<GetStatusQservHttpMgtRequest> const&)> CallbackType;
+
+    GetStatusQservHttpMgtRequest() = delete;
+    GetStatusQservHttpMgtRequest(GetStatusQservHttpMgtRequest const&) = delete;
+    GetStatusQservHttpMgtRequest& operator=(GetStatusQservHttpMgtRequest const&) = delete;
+
+    ~GetStatusQservHttpMgtRequest() final = default;
+
+    /**
+     * Static factory method is needed to prevent issues with the lifespan
+     * and memory management of instances created otherwise (as values or via
+     * low-level pointers).
+     * @param serviceProvider A reference to a provider of services for accessing
+     *   Configuration, saving the request's persistent state to the database.
+     * @param worker The name of a worker to send the request to.
+     * @param taskSelector (optional) task selection criterias
+     * @param onFinish (optional) callback function to be called upon request completion.
+     * @return A pointer to the created object.
+     */
+    static std::shared_ptr<GetStatusQservHttpMgtRequest> create(
+            std::shared_ptr<ServiceProvider> const& serviceProvider, std::string const& worker,
+            wbase::TaskSelector const& taskSelector = wbase::TaskSelector(),
+            CallbackType const& onFinish = nullptr);
+
+    /**
+     * @return The info object returned by the worker.
+     * @throw std::logic_error if called before the request finishes or if it failed.
+     */
+    nlohmann::json const& info() const;
+
+protected:
+    /// @see QservHttpMgtRequest::createHttpReq()
+    void createHttpReqImpl(replica::Lock const& lock) final;
+
+    /// @see QservHttpMgtRequest::notify()
+    void notify(replica::Lock const& lock) final;
+
+private:
+    /// @see GetStatusQservHttpMgtRequest::create()
+    GetStatusQservHttpMgtRequest(std::shared_ptr<ServiceProvider> const& serviceProvider,
+                                 std::string const& worker, wbase::TaskSelector const& taskSelector,
+                                 CallbackType const& onFinish);
+
+    /**
+     * Extract and process data of the completed request.
+     * @param The HTTP request to be analyzed.
+     */
+    void _processResponse(std::shared_ptr<http::AsyncReq> const& request);
+
+    // Input parameters
+
+    wbase::TaskSelector const _taskSelector;
+    CallbackType _onFinish;  ///< this object is reset after finishing the request
+
+    /// The info object returned by the Qserv worker
+    nlohmann::json _info;
+};
+
+}  // namespace lsst::qserv::replica
+
+#endif  // LSST_QSERV_REPLICA_GETSTATUSQSERVHTTPMGTREQUEST_H