Merge branch 'tickets/DM-55404'

Author: iagaponenko

python/lsst/qserv/admin/replication_interface.py

@@ -43,7 +43,7 @@
 
 chunk_info_file = "chunk_info.json"
 
-repl_api_version = 56
+repl_api_version = 57
 
 _log = logging.getLogger(__name__)
 

src/cconfig/CMakeLists.txt

@@ -2,6 +2,8 @@ add_library(cconfig SHARED)
 
 target_sources(cconfig PRIVATE
     CzarConfig.cc
+    DataManagementEvent.cc
+    EventService.cc
 )
 
 target_include_directories(cconfig PRIVATE
@@ -11,8 +13,27 @@ target_include_directories(cconfig PRIVATE
 target_link_libraries(cconfig PUBLIC
     log
     XrdSsiLib
+    boost_system
+    util
 )
 
 install(
     TARGETS cconfig
 )
+
+FUNCTION(cconfig_tests)
+    FOREACH(TEST IN ITEMS ${ARGV})
+        add_executable(${TEST} ${TEST}.cc)
+        target_link_libraries(${TEST} PUBLIC
+            cconfig
+            Boost::unit_test_framework
+            Threads::Threads
+            mysql
+        )
+        add_test(NAME ${TEST} COMMAND ${TEST})
+    ENDFOREACH()
+ENDFUNCTION()
+
+cconfig_tests(
+    testEventService
+)

src/cconfig/CzarConfig.cc

@@ -111,6 +111,10 @@ CzarConfig::CzarConfig(util::ConfigStore const& configStore, std::string const&
         throw util::ConfigException(ERR_LOC, " CzarConfig::" + std::string(__func__) +
                                                      ": 'replication.num_http_threads' can't be 0.");
     }
+    if (_replicationNumEventThreads->getVal() == 0) {
+        throw util::ConfigException(ERR_LOC, " CzarConfig::" + std::string(__func__) +
+                                                     ": 'replication.num_event_threads' can't be 0.");
+    }
 
     // Cache the cached version of the configuration in the JSON format. The JSON object
     // contains two collections of parameters: the "input" ones that were passed into

src/cconfig/CzarConfig.h

@@ -206,6 +206,7 @@ class CzarConfig {
     }
     uint16_t replicationHttpPort() const { return _replicationHttpPort->getVal(); }
     size_t replicationNumHttpThreads() const { return _replicationNumHttpThreads->getVal(); }
+    size_t replicationNumEventThreads() const { return _replicationNumEventThreads->getVal(); }
     std::string const& httpUser() const { return _httpUser->getVal(); }
     void setHttpUser(std::string const& user);
     std::string const& httpPassword() const { return _httpPassword->getVal(); }
@@ -386,6 +387,8 @@ class CzarConfig {
             util::ConfigValTInt::create(_configValMap, "replication", "http_port", notReq, 0);
     CVTUIntPtr _replicationNumHttpThreads =
             util::ConfigValTUInt::create(_configValMap, "replication", "num_http_threads", notReq, 2);
+    CVTUIntPtr _replicationNumEventThreads =
+            util::ConfigValTUInt::create(_configValMap, "replication", "num_event_threads", notReq, 2);
 
     // User and password for the HTTP frontend
     CVTStrPtr _httpUser = util::ConfigValTStr::create(_configValMap, "http", "user", notReq, "");

src/cconfig/DataManagementEvent.cc

@@ -0,0 +1,114 @@
+/*
+ * 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/>.
+ */
+
+#include "cconfig/DataManagementEvent.h"
+
+// System headers
+#include <stdexcept>
+
+using namespace std;
+using json = nlohmann::json;
+
+namespace {
+
+string parseDatabase(json const& jsonObj, string const& context) {
+    if (!jsonObj.contains("database")) {
+        throw invalid_argument(context + ": Missing required field 'database'");
+    }
+    return jsonObj.at("database").get<string>();
+}
+
+string parseTable(json const& jsonObj, string const& context) {
+    if (!jsonObj.contains("table")) {
+        throw invalid_argument(context + ": Missing required field 'table'");
+    }
+    return jsonObj.at("table").get<string>();
+}
+
+}  // namespace
+
+namespace lsst::qserv::cconfig {
+
+string DataManagementEvent::type2str(Type type) {
+    switch (type) {
+        case Type::NONE:
+            return "NONE";
+        case Type::CHUNK_MAP_REBUILT:
+            return "CHUNK_MAP_REBUILT";
+        case Type::DATABASE_PUBLISHED:
+            return "DATABASE_PUBLISHED";
+        case Type::DATABASE_DELETED:
+            return "DATABASE_DELETED";
+        case Type::TABLE_DELETED:
+            return "TABLE_DELETED";
+    }
+    throw invalid_argument("Unknown DataManagementEvent::Type: " + std::to_string(static_cast<int>(type)));
+}
+
+DataManagementEvent::Type DataManagementEvent::str2type(string const& str) {
+    string const context = "DataManagementEvent::" + string(__func__);
+    if (str == "NONE") return Type::NONE;
+    if (str == "CHUNK_MAP_REBUILT") return Type::CHUNK_MAP_REBUILT;
+    if (str == "DATABASE_PUBLISHED") return Type::DATABASE_PUBLISHED;
+    if (str == "DATABASE_DELETED") return Type::DATABASE_DELETED;
+    if (str == "TABLE_DELETED") return Type::TABLE_DELETED;
+    throw invalid_argument(context + ": unknown DataManagementEvent::Type: '" + str + "'");
+}
+
+DataManagementEvent DataManagementEvent::fromJson(json const& jsonObj) {
+    string const context = "DataManagementEvent::" + string(__func__);
+    DataManagementEvent event;
+    if (!jsonObj.is_object()) throw invalid_argument(context + ": JSON object expected");
+    if (!jsonObj.contains("timestamp"))
+        throw invalid_argument(context + ": Missing required field 'timestamp'");
+    event.timestamp =
+            chrono::system_clock::time_point(chrono::milliseconds(jsonObj.at("timestamp").get<long long>()));
+    if (!jsonObj.contains("type")) throw invalid_argument(context + ": Missing required field 'type'");
+    event.type = DataManagementEvent::str2type(jsonObj.at("type").get<string>());
+    switch (event.type) {
+        case Type::NONE:
+            break;
+        case Type::CHUNK_MAP_REBUILT:
+            break;
+        case Type::DATABASE_PUBLISHED:
+            event.database = ::parseDatabase(jsonObj, context);
+            break;
+        case Type::DATABASE_DELETED:
+            event.database = ::parseDatabase(jsonObj, context);
+            break;
+        case Type::TABLE_DELETED:
+            event.database = ::parseDatabase(jsonObj, context);
+            event.table = ::parseTable(jsonObj, context);
+            break;
+        default:
+            throw invalid_argument(context + ": Unknown DataManagementEvent::Type");
+    }
+    return event;
+}
+
+json DataManagementEvent::toJson() const {
+    return {{"timestamp", chrono::duration_cast<chrono::milliseconds>(timestamp.time_since_epoch()).count()},
+            {"type", DataManagementEvent::type2str(type)},
+            {"database", database},
+            {"table", table}};
+}
+
+}  // namespace lsst::qserv::cconfig

src/cconfig/DataManagementEvent.h

@@ -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/>.
+ */
+#ifndef LSST_QSERV_CCONFIG_DATAMANAGEMENTEVENT_H
+#define LSST_QSERV_CCONFIG_DATAMANAGEMENTEVENT_H
+
+// System headers
+#include <chrono>
+#include <string>
+
+// Third party headers
+#include "nlohmann/json.hpp"
+
+// This header declarations
+namespace lsst::qserv::cconfig {
+
+/**
+ * Class DataManagementEvent is an abstraction for the events posted by the Replication system.
+ * It contains information about the event type and any associated data.
+ */
+class DataManagementEvent {
+public:
+    enum class Type : int {
+        NONE = 0,
+        CHUNK_MAP_REBUILT,
+        DATABASE_PUBLISHED,
+        DATABASE_DELETED,
+        TABLE_DELETED
+    };
+
+    static std::string type2str(Type type);
+    static Type str2type(std::string const& str);
+
+    /**
+     * Construct a DataManagementEvent object from a JSON representation.
+     *
+     * @param jsonObj The JSON object containing the event data.
+     * @return A DataManagementEvent object.
+     * @throws std::invalid_argument if the JSON object is missing required fields or contains invalid data.
+     */
+    static DataManagementEvent fromJson(nlohmann::json const& jsonObj);
+
+    nlohmann::json toJson() const;
+
+    // Timestamp of the event.
+    std::chrono::system_clock::time_point timestamp = std::chrono::system_clock::now();
+
+    // Mandatory attribute
+    Type type = Type::NONE;
+
+    // Optional attributes which depend on the event type.
+    std::string database;
+    std::string table;
+};
+
+}  // namespace lsst::qserv::cconfig
+
+#endif  // LSST_QSERV_CCONFIG_DATAMANAGEMENTEVENT_H