Data management event handling services at the Replication system

Author: iagaponenko

src/replica/CMakeLists.txt

@@ -23,6 +23,7 @@ target_link_libraries(replica PUBLIC
     replica_util
     replica_worker
     css
+    cconfig
     xrdreq
     xrdsvc
     XrdCl

src/replica/contr/HttpIngestModule.cc

@@ -35,6 +35,7 @@
 #include "boost/algorithm/string.hpp"
 
 // Qserv headers
+#include "cconfig/DataManagementEvent.h"
 #include "css/CssAccess.h"
 #include "css/CssError.h"
 #include "css/DbInterfaceMySql.h"
@@ -53,7 +54,10 @@
 #include "replica/jobs/SqlGrantAccessJob.h"
 #include "replica/jobs/SqlEnableDbJob.h"
 #include "replica/jobs/SqlRemoveTablePartitionsJob.h"
+#include "replica/qserv/PostEventQservCzarMgtRequest.h"
+#include "replica/registry/Registry.h"
 #include "replica/requests/SqlResultSet.h"
+#include "replica/services/ChunkMap.h"
 #include "replica/services/DatabaseServices.h"
 #include "replica/services/ServiceProvider.h"
 #include "replica/util/ChunkedTable.h"
@@ -377,6 +381,12 @@ json HttpIngestModule::_publishDatabase() {
     // the Replication system.
     _qservSync(database, allWorkers);
 
+    // Notify Czars about the database publication event.
+    cconfig::DataManagementEvent dataManagementEvent;
+    dataManagementEvent.type = cconfig::DataManagementEvent::Type::DATABASE_PUBLISHED;
+    dataManagementEvent.database = database.name;
+    _notifyCzars(dataManagementEvent);
+
     ControllerEvent event;
     event.status = "PUBLISH DATABASE";
     event.kvInfo.emplace_back("database", database.name);
@@ -481,6 +491,12 @@ json HttpIngestModule::_deleteDatabase() {
     error = reconfigureWorkers(database, allWorkers, workerReconfigTimeoutSec());
     if (!error.empty()) throw http::Error(__func__, error);
 
+    // Notify Czars about the database removal event.
+    cconfig::DataManagementEvent dataManagementEvent;
+    dataManagementEvent.type = cconfig::DataManagementEvent::Type::DATABASE_DELETED;
+    dataManagementEvent.database = database.name;
+    _notifyCzars(dataManagementEvent);
+
     return json::object();
 }
 
@@ -745,6 +761,13 @@ json HttpIngestModule::_deleteTable() {
     error = reconfigureWorkers(database, allWorkers, workerReconfigTimeoutSec());
     if (!error.empty()) throw http::Error(__func__, error);
 
+    // Notify Czars about the table removal event.
+    cconfig::DataManagementEvent dataManagementEvent;
+    dataManagementEvent.type = cconfig::DataManagementEvent::Type::TABLE_DELETED;
+    dataManagementEvent.database = database.name;
+    dataManagementEvent.table = table.name;
+    _notifyCzars(dataManagementEvent);
+
     return json::object();
 }
 
@@ -1414,4 +1437,17 @@ void HttpIngestModule::_qservSync(DatabaseInfo const& database, bool allWorkers)
     }
 }
 
+void HttpIngestModule::_notifyCzars(cconfig::DataManagementEvent const& dataManagementEvent) const {
+    // The replica disposition map neeeds to be updated in the database before sending
+    // the notification to the czars.
+    auto const serviceProvider = controller()->serviceProvider();
+    serviceProvider->chunkMap()->update();
+    auto const czars = serviceProvider->registry()->czars();
+    for (auto const& czar : czars) {
+        auto const request =
+                PostEventQservCzarMgtRequest::create(serviceProvider, czar.name, dataManagementEvent);
+        request->start();
+    }
+}
+
 }  // namespace lsst::qserv::replica

src/replica/contr/HttpIngestModule.h

@@ -34,6 +34,10 @@
 #include "replica/util/Common.h"
 
 // Forward declarations
+namespace lsst::qserv::cconfig {
+class DataManagementEvent;
+}  // namespace lsst::qserv::cconfig
+
 namespace lsst::qserv::replica {
 class DatabaseInfo;
 }  // namespace lsst::qserv::replica
@@ -182,7 +186,6 @@ class HttpIngestModule : public HttpModule {
     /**
      * Grant SELECT authorizations for the new database to Qserv
      * MySQL account(s) at workers.
-     *
      * @param database database descriptor
      * @param allWorkers  'true' if all workers should be involved into the operation
      * @throws http::Error if the operation failed
@@ -192,7 +195,6 @@ class HttpIngestModule : public HttpModule {
     /**
      * Enable this database in Qserv workers by adding an entry
      * to table 'qservw_worker.Dbs' at workers.
-     *
      * @param database database descriptor
      * @param allWorkers 'true' if all workers should be involved into the operation
      * @throws http::Error if the operation failed
@@ -205,7 +207,6 @@ class HttpIngestModule : public HttpModule {
      * tables have chunks representations for all registered chunks, even though some
      * of the chunks may be empty. This stage enforces structural consistency across
      * partitioned tables.
-     *
      * @param database database descriptor
      * @param allWorkers 'true' if all workers should be involved into the operation
      * @throws http::Error if the operation failed
@@ -214,7 +215,6 @@ class HttpIngestModule : public HttpModule {
 
     /**
      * Consolidate MySQL partitioned tables at workers by removing partitions.
-     *
      * @param database database descriptor
      * @param allWorkers 'true' if all workers should be involved into the operation
      * @throws http::Error if operation failed
@@ -235,7 +235,6 @@ class HttpIngestModule : public HttpModule {
     /**
      * (Re-)build the empty chunks list (table) for the specified database.
      * The methods throws exceptions in case of any errors.
-     *
      * @param databaseName The name of a database.
      * @param force Rebuild the list if 'true'.
      * @return An object representing a result of the operation (empty chunk list
@@ -266,13 +265,18 @@ class HttpIngestModule : public HttpModule {
      * It runs the Replication system's chunks scanner to register chunk info
      * in the persistent state of the system. It also registers (synchronizes)
      * new chunks at Qserv workers.
-     *
      * @param database database descriptor
      * @param allWorkers 'true' if all workers should be involved into the operation
      * @throws http::Error if the operation failed
      */
     void _qservSync(DatabaseInfo const& database, bool allWorkers) const;
 
+    /**
+     * Notify Czars about a data management event.
+     * @param dataManagementEvent the event to be posted to Czars
+     */
+    void _notifyCzars(cconfig::DataManagementEvent const& dataManagementEvent) const;
+
     // The name and a type of a special column for the super-transaction-based ingest
 
     static std::string const _partitionByColumn;

src/replica/contr/ReplicationTask.cc

@@ -56,7 +56,7 @@ bool ReplicationTask::onRun() {
     launch<FindAllJob>(priority, saveReplicaInfo, allWorkers);
     if (!_disableQservSync) sync(_qservSyncTimeoutSec, _forceQservSync);
 
-    if (_qservChunkMapUpdate) _updateChunkMap();
+    if (_qservChunkMapUpdate) serviceProvider()->chunkMap()->update();
 
     launch<FixUpJob>(priority);
     if (!_disableQservSync) sync(_qservSyncTimeoutSec, _forceQservSync);
@@ -88,12 +88,4 @@ ReplicationTask::ReplicationTask(Controller::Ptr const& controller,
           _qservChunkMapUpdate(qservChunkMapUpdate),
           _purge(purge) {}
 
-void ReplicationTask::_updateChunkMap() {
-    if (!serviceProvider()->chunkMap()->update()) {
-        error("failed to update chunk map in the Czar database");
-    } else {
-        info("chunk map has been updated in the Czar database");
-    }
-}
-
 }  // namespace lsst::qserv::replica

src/replica/contr/ReplicationTask.h

@@ -81,9 +81,6 @@ class ReplicationTask : public Task {
                     unsigned int qservSyncTimeoutSec, bool disableQservSync, bool forceQservSync,
                     bool qservChunkMapUpdate, unsigned int replicationIntervalSec, bool purge);
 
-    /// Update the chunk disposition map in QMeta when changes in the map are detected.
-    void _updateChunkMap();
-
     /// The maximum number of seconds to be waited before giving up
     /// on the Qserv synchronization requests.
     unsigned int const _qservSyncTimeoutSec;

src/replica/qserv/CMakeLists.txt

@@ -16,6 +16,7 @@ target_sources(replica_qserv PRIVATE
     QservMgtRequest.cc
     QservMgtServices.cc
     QservWorkerMgtRequest.cc
+    PostEventQservCzarMgtRequest.cc
     RemoveReplicaQservMgtRequest.cc
     SetReplicasQservMgtRequest.cc
     TestEchoQservMgtRequest.cc

src/replica/qserv/PostEventQservCzarMgtRequest.cc

@@ -0,0 +1,75 @@
+/*
+ * 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/qserv/PostEventQservCzarMgtRequest.h"
+
+// Qserv headers
+#include "http/Method.h"
+#include "replica/util/Common.h"
+
+// LSST headers
+#include "lsst/log/Log.h"
+
+using namespace nlohmann;
+using namespace std;
+
+namespace {
+
+LOG_LOGGER _log = LOG_GET("lsst.qserv.replica.PostEventQservCzarMgtRequest");
+
+}  // namespace
+
+namespace lsst::qserv::replica {
+
+PostEventQservCzarMgtRequest::Ptr PostEventQservCzarMgtRequest::create(
+        shared_ptr<ServiceProvider> const& serviceProvider, string const& czarName,
+        cconfig::DataManagementEvent const& event,
+        PostEventQservCzarMgtRequest::CallbackType const& onFinish) {
+    return PostEventQservCzarMgtRequest::Ptr(
+            new PostEventQservCzarMgtRequest(serviceProvider, czarName, event, onFinish));
+}
+
+PostEventQservCzarMgtRequest::PostEventQservCzarMgtRequest(
+        shared_ptr<ServiceProvider> const& serviceProvider, string const& czarName,
+        cconfig::DataManagementEvent const& event, PostEventQservCzarMgtRequest::CallbackType const& onFinish)
+        : QservCzarMgtRequest(serviceProvider, "QSERV_CZAR_POST_EVENT", czarName),
+          _event(event),
+          _onFinish(onFinish) {}
+
+list<pair<string, string>> PostEventQservCzarMgtRequest::extendedPersistentState() const {
+    list<pair<string, string>> result;
+    result.emplace_back("czar", czarName());
+    return result;
+}
+
+void PostEventQservCzarMgtRequest::createHttpReqImpl(replica::Lock const& lock) {
+    string const target = "/event";
+    json const data = json::object({{"czar", czarName()}, {"event", _event.toJson()}});
+    createHttpReq(lock, http::Method::POST, target, data);
+}
+
+void PostEventQservCzarMgtRequest::notify(replica::Lock const& lock) {
+    LOGS(_log, LOG_LVL_TRACE, context() << __func__);
+    notifyDefaultImpl<PostEventQservCzarMgtRequest>(lock, _onFinish);
+}
+
+}  // namespace lsst::qserv::replica

src/replica/qserv/PostEventQservCzarMgtRequest.h

@@ -0,0 +1,99 @@
+/*
+ * 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_POSTEVENTQSERVCZARMGTREQUEST_H
+#define LSST_QSERV_REPLICA_POSTEVENTQSERVCZARMGTREQUEST_H
+
+// System headers
+#include <list>
+#include <memory>
+#include <string>
+#include <utility>
+
+// Third party headers
+#include "nlohmann/json.hpp"
+
+// Qserv headers
+#include "cconfig/DataManagementEvent.h"
+#include "replica/qserv/QservCzarMgtRequest.h"
+
+namespace lsst::qserv::replica {
+class ServiceProvider;
+}  // namespace lsst::qserv::replica
+
+// This header declarations
+namespace lsst::qserv::replica {
+
+/**
+ * Class PostEventQservCzarMgtRequest implements a request for posting the data
+ * management events to the Qserv Czar.
+ */
+class PostEventQservCzarMgtRequest : public QservCzarMgtRequest {
+public:
+    typedef std::shared_ptr<PostEventQservCzarMgtRequest> Ptr;
+
+    /// The function type for notifications on the completion of the request
+    typedef std::function<void(Ptr)> CallbackType;
+
+    PostEventQservCzarMgtRequest() = delete;
+    PostEventQservCzarMgtRequest(PostEventQservCzarMgtRequest const&) = delete;
+    PostEventQservCzarMgtRequest& operator=(PostEventQservCzarMgtRequest const&) = delete;
+
+    virtual ~PostEventQservCzarMgtRequest() override = 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 czarName The name of a Czar to send the request to.
+     * @param event The management event to be posted to the Czar.
+     * @param onFinish (optional) callback function to be called upon request completion.
+     * @return A pointer to the created object.
+     */
+    static Ptr create(ServiceProvider::Ptr const& serviceProvider, std::string const& czarName,
+                      cconfig::DataManagementEvent const& event, CallbackType const& onFinish = nullptr);
+
+    /// @see QservMgtRequest::extendedPersistentState()
+    std::list<std::pair<std::string, std::string>> extendedPersistentState() const override;
+
+protected:
+    /// @see QservMgtRequest::createHttpReqImpl()
+    virtual void createHttpReqImpl(replica::Lock const& lock) override;
+
+    /// @see QservMgtRequest::notify
+    virtual void notify(replica::Lock const& lock) override;
+
+private:
+    /// @see PostEventQservCzarMgtRequest::create()
+    PostEventQservCzarMgtRequest(ServiceProvider::Ptr const& serviceProvider, std::string const& czarName,
+                                 cconfig::DataManagementEvent const& event, CallbackType const& onFinish);
+
+    // Input parameters
+
+    cconfig::DataManagementEvent const _event;  ///< The management event to be posted to the Czar.
+    CallbackType _onFinish;                     ///< The callback function is reset when the request finishes.
+};
+
+}  // namespace lsst::qserv::replica
+
+#endif  // LSST_QSERV_REPLICA_POSTEVENTQSERVCZARMGTREQUEST_H

src/replica/services/ChunkMap.h

@@ -95,7 +95,7 @@ class ChunkMap : public std::enable_shared_from_this<ChunkMap> {
 
     // The internal representation of the current state of the chunk map.
 
-    std::shared_ptr<Chunks> _chunks;  ///< The the current state.
+    std::shared_ptr<Chunks> _chunks;  ///< The current state.
     mutable std::mutex _mtx;          ///< Mutex to protect access to the chunk map.
 };
 

src/replica/services/ServiceProvider.h

@@ -33,12 +33,12 @@
 // Qserv headers
 #include "http/Auth.h"
 #include "replica/services/ChunkLocker.h"
-#include "replica/services/ChunkMap.h"
 #include "replica/util/Mutex.h"
 #include "replica/util/NamedMutexRegistry.h"
 
 // Forward declarations
 namespace lsst::qserv::replica {
+class ChunkMap;
 class Configuration;
 class DatabaseServices;
 class Messenger;