[FLASH-739] Add cluster manager service by poco timer

dbms/src/Server/CMakeLists.txt

@@ -24,7 +24,8 @@ add_library (clickhouse-server-lib
     RootRequestHandler.cpp
     Server.cpp
     StatusFile.cpp
-    TCPHandler.cpp)
+    TCPHandler.cpp
+    ClusterManagerService.cpp)
 
 target_link_libraries (clickhouse-server-lib clickhouse_common_io daemon clickhouse_storages_system clickhouse_functions clickhouse_aggregate_functions clickhouse_table_functions)
 target_include_directories (clickhouse-server-lib PUBLIC ${ClickHouse_SOURCE_DIR}/libs/libdaemon/include)

dbms/src/Server/ClusterManagerService.cpp

@@ -0,0 +1,49 @@
+#include "ClusterManagerService.h"
+
+
+#include <Common/FunctionTimerTask.h>
+#include <Common/ShellCommand.h>
+#include <Interpreters/Context.h>
+#include <Poco/File.h>
+#include <Poco/Path.h>
+#include <Storages/Transaction/TMTContext.h>
+#include <common/logger_useful.h>
+
+namespace DB
+{
+
+const std::string CLUSTER_MANAGER_PATH_KEY = "flash.flash_cluster.cluster_manager_path";
+const std::string BIN_NAME = "flash_cluster_manager";
+const std::string TASK_INTERVAL_KEY = "flash.flash_cluster.update_rule_interval";
+
+constexpr long SECOND = 1000;
+constexpr long INIT_DELAY = SECOND * 5;
+
+void ClusterManagerService::run(const std::string & bin_path, const std::vector<std::string> & args)
+{
+    auto proc = ShellCommand::executeDirect(bin_path, args);
+    proc->wait();
+}
+
+ClusterManagerService::ClusterManagerService(DB::Context & context_, const std::string config_path)
+    : context(context_), timer(), log(&Logger::get("ClusterManagerService"))
+{
+    const auto & conf = context.getConfigRef();
+
+    auto bin_path = conf.getString(CLUSTER_MANAGER_PATH_KEY, ".") + Poco::Path::separator() + BIN_NAME;
+    auto task_interval = conf.getInt(TASK_INTERVAL_KEY, 10) * SECOND;
+
+    if (!Poco::File(bin_path).exists())
+        throw Exception("Cluster manager binary file does not exist: " + bin_path);
+
+    std::vector<std::string> args;
+    args.push_back(config_path);
+
+    LOG_INFO(log, "Registered timed cluster manager task at rate " << task_interval / SECOND << " seconds");
+
+    timer.scheduleAtFixedRate(FunctionTimerTask::create(std::bind(&ClusterManagerService::run, bin_path, args)), INIT_DELAY, task_interval);
+}
+
+ClusterManagerService::~ClusterManagerService() { timer.cancel(true); }
+
+} // namespace DB

dbms/src/Server/ClusterManagerService.h

@@ -0,0 +1,27 @@
+#pragma once
+
+#include <Poco/Util/Timer.h>
+#include <Storages/MergeTree/BackgroundProcessingPool.h>
+#include <common/logger_useful.h>
+
+namespace DB
+{
+
+class Context;
+class BackgroundProcessingPool;
+
+class ClusterManagerService : private boost::noncopyable
+{
+public:
+    ClusterManagerService(Context & context_, const std::string config_path);
+    ~ClusterManagerService();
+
+private:
+    static void run(const std::string & command, const std::vector<std::string> & args);
+    Context & context;
+    Poco::Util::Timer timer;
+    Logger * log;
+};
+
+
+} // namespace DB

dbms/src/Server/Server.cpp

@@ -40,6 +40,7 @@
 #include "MetricsTransmitter.h"
 #include "StatusFile.h"
 #include "TCPHandlerFactory.h"
+#include "ClusterManagerService.h"
 
 #if Poco_NetSSL_FOUND
 #include <Poco/Net/Context.h>
@@ -783,6 +784,7 @@ int Server::main(const std::vector<std::string> & /*args*/)
         }
 
         SessionCleaner session_cleaner(*global_context);
+        ClusterManagerService clusterMgrSvc(*global_context, config_path);
 
         waitForTerminationRequest();
     }