ClickHouse · alexey-milovidov · Aug 15, 2019 · Aug 12, 2019 · Aug 12, 2019 · Aug 12, 2019
diff --git a/dbms/programs/server/config.d/metric_log.xml b/dbms/programs/server/config.d/metric_log.xml
@@ -0,0 +1,8 @@
+<yandex>
+    <metric_log>
+        <database>system</database>
+        <table>metric_log</table>
+        <flush_interval_milliseconds>7500</flush_interval_milliseconds>
+        <collect_interval_milliseconds>1000</collect_interval_milliseconds>
+    </metric_log>
+</yandex>
diff --git a/dbms/programs/server/config.xml b/dbms/programs/server/config.xml
@@ -331,6 +331,16 @@
     </text_log>
     -->
 
+    <!-- Uncomment to write metric log into table.
+         Metric log contains rows with current values of ProfileEvents, CurrentMetrics collected with "collect_interval_milliseconds" interval.
+    <metric_log>
+        <database>system</database>
+        <table>metric_log</table>
+        <flush_interval_milliseconds>7500</flush_interval_milliseconds>
+        <collect_interval_milliseconds>1000</collect_interval_milliseconds>
+    </metric_log>
+    -->
+
     <!-- Parameters for embedded dictionaries, used in Yandex.Metrica.
          See https://clickhouse.yandex/docs/en/dicts/internal_dicts/
     -->

diff --git a/dbms/src/Interpreters/Context.cpp b/dbms/src/Interpreters/Context.cpp
@@ -38,10 +38,6 @@
 #include <Interpreters/Compiler.h>
 #include <Interpreters/SettingsConstraints.h>
 #include <Interpreters/SystemLog.h>
-#include <Interpreters/QueryLog.h>
-#include <Interpreters/QueryThreadLog.h>
-#include <Interpreters/PartLog.h>
-#include <Interpreters/TraceLog.h>
 #include <Interpreters/Context.h>
 #include <Interpreters/DDLWorker.h>
 #include <Common/DNSResolver.h>
@@ -1701,6 +1697,7 @@ std::shared_ptr<PartLog> Context::getPartLog(const String & part_database)
     return shared->system_logs->part_log;
 }
 
+
 std::shared_ptr<TraceLog> Context::getTraceLog()
 {
     auto lock = getLock();
@@ -1711,6 +1708,7 @@ std::shared_ptr<TraceLog> Context::getTraceLog()
     return shared->system_logs->trace_log;
 }
 
+
 std::shared_ptr<TextLog> Context::getTextLog()
 {
     auto lock = getLock();
@@ -1722,6 +1720,17 @@ std::shared_ptr<TextLog> Context::getTextLog()
 }
 
 
+std::shared_ptr<MetricLog> Context::getMetricLog()
+{
+    auto lock = getLock();
+
+    if (!shared->system_logs || !shared->system_logs->metric_log)
+        return {};
+
+    return shared->system_logs->metric_log;
+}
+
+
 CompressionCodecPtr Context::chooseCompressionCodec(size_t part_size, double part_size_ratio) const
 {
     auto lock = getLock();

diff --git a/dbms/src/Interpreters/Context.h b/dbms/src/Interpreters/Context.h
@@ -64,6 +64,7 @@ class QueryThreadLog;
 class PartLog;
 class TextLog;
 class TraceLog;
+class MetricLog;
 struct MergeTreeSettings;
 class IDatabase;
 class DDLGuard;
@@ -434,6 +435,7 @@ class Context
     std::shared_ptr<QueryThreadLog> getQueryThreadLog();
     std::shared_ptr<TraceLog> getTraceLog();
     std::shared_ptr<TextLog> getTextLog();
+    std::shared_ptr<MetricLog> getMetricLog();
 
     /// Returns an object used to log opertaions with parts if it possible.
     /// Provide table name to make required cheks.

diff --git a/dbms/src/Interpreters/InterpreterSystemQuery.cpp b/dbms/src/Interpreters/InterpreterSystemQuery.cpp
@@ -16,6 +16,7 @@
 #include <Interpreters/QueryThreadLog.h>
 #include <Interpreters/TraceLog.h>
 #include <Interpreters/TextLog.h>
+#include <Interpreters/MetricLog.h>
 #include <Databases/IDatabase.h>
 #include <Storages/StorageDistributed.h>
 #include <Storages/StorageReplicatedMergeTree.h>
@@ -232,7 +233,8 @@ BlockIO InterpreterSystemQuery::execute()
                     [&] () { if (auto part_log = context.getPartLog("")) part_log->flush(); },
                     [&] () { if (auto query_thread_log = context.getQueryThreadLog()) query_thread_log->flush(); },
                     [&] () { if (auto trace_log = context.getTraceLog()) trace_log->flush(); },
-                    [&] () { if (auto text_log = context.getTextLog()) text_log->flush(); }
+                    [&] () { if (auto text_log = context.getTextLog()) text_log->flush(); },
+                    [&] () { if (auto metric_log = context.getMetricLog()) metric_log->flush(); }
             );
             break;
         case Type::STOP_LISTEN_QUERIES:

diff --git a/dbms/src/Interpreters/MetricLog.cpp b/dbms/src/Interpreters/MetricLog.cpp
@@ -0,0 +1,118 @@
+#include <Interpreters/MetricLog.h>
+#include <DataTypes/DataTypesNumber.h>
+#include <DataTypes/DataTypeString.h>
+#include <DataTypes/DataTypeDate.h>
+#include <DataTypes/DataTypeDateTime.h>
+
+namespace DB
+{
+
+Block MetricLogElement::createBlock()
+{
+    ColumnsWithTypeAndName columns_with_type_and_name;
+
+    columns_with_type_and_name.emplace_back(std::make_shared<DataTypeDate>(),       "event_date");
+    columns_with_type_and_name.emplace_back(std::make_shared<DataTypeDateTime>(),   "event_time");
+    columns_with_type_and_name.emplace_back(std::make_shared<DataTypeUInt64>(),     "milliseconds");
+
+    //ProfileEvents
+    for (size_t i = 0, end = ProfileEvents::end(); i < end; ++i)
+    {
+        std::string name;
+        name += "ProfileEvent_";
+        name += ProfileEvents::getName(ProfileEvents::Event(i));
+        columns_with_type_and_name.emplace_back(std::make_shared<DataTypeUInt64>(), std::move(name));
+    }
+
+    //CurrentMetrics
+    for (size_t i = 0, end = CurrentMetrics::end(); i < end; ++i)
+    {
+        std::string name;
+        name += "CurrentMetric_";
+        name += CurrentMetrics::getName(ProfileEvents::Event(i));
+        columns_with_type_and_name.emplace_back(std::make_shared<DataTypeInt64>(), std::move(name));
+    }
+
+    return Block(columns_with_type_and_name);
+}
+
+void MetricLogElement::appendToBlock(Block & block) const
+{
+    MutableColumns columns = block.mutateColumns();
+
+    size_t iter = 0;
+
+    columns[iter++]->insert(DateLUT::instance().toDayNum(event_time));
+    columns[iter++]->insert(event_time);
+    columns[iter++]->insert(milliseconds);
+
+    //ProfileEvents
+    for (size_t i = 0, end = ProfileEvents::end(); i < end; ++i)
+    {
+        const UInt64 value = ProfileEvents::global_counters[i].load(std::memory_order_relaxed);
+        columns[iter++]->insert(value);
+    }
+
+    //CurrentMetrics
+    for (size_t i = 0, end = CurrentMetrics::end(); i < end; ++i)
+    {
+        const UInt64 value = CurrentMetrics::values[i];
+        columns[iter++]->insert(value);
+    }
+}
+
+void MetricLog::startCollectMetric(size_t collect_interval_milliseconds_)
+{
+    collect_interval_milliseconds = collect_interval_milliseconds_;
+    is_shutdown_metric_thread = false;
+    metric_flush_thread = ThreadFromGlobalPool([this] { metricThreadFunction(); });
+}
+
+void MetricLog::stopCollectMetric()
+{
+    bool old_val = false;
+    if (!is_shutdown_metric_thread.compare_exchange_strong(old_val, true))
+        return;
+    metric_flush_thread.join();
+}
+
+inline UInt64 time_in_milliseconds(std::chrono::time_point<std::chrono::system_clock> timepoint)
+{
+    return std::chrono::duration_cast<std::chrono::milliseconds>(timepoint.time_since_epoch()).count();
+}
+
+inline UInt64 time_in_seconds(std::chrono::time_point<std::chrono::system_clock> timepoint)
+{
+    return std::chrono::duration_cast<std::chrono::seconds>(timepoint.time_since_epoch()).count();
+}
+
+void MetricLog::metricThreadFunction()
+{
+    auto desired_timepoint = std::chrono::system_clock::now();
+    while (true)
+    {
+        try
+        {
+            if (is_shutdown_metric_thread)
+                break;
+
+            MetricLogElement elem;
+            const auto prev_timepoint = std::chrono::system_clock::now();
+            elem.event_time = std::chrono::system_clock::to_time_t(prev_timepoint);
+            elem.milliseconds = time_in_milliseconds(prev_timepoint) - time_in_seconds(prev_timepoint) * 1000;
+
+            this->add(elem);
+
+            while (desired_timepoint <= std::chrono::system_clock::now())
+                desired_timepoint += std::chrono::milliseconds(collect_interval_milliseconds);
+            std::this_thread::sleep_until(desired_timepoint);
+        }
+        catch (...)
+        {
+            tryLogCurrentException(__PRETTY_FUNCTION__);
+        }
+
+    }
+}
+
+}
diff --git a/dbms/src/Interpreters/MetricLog.h b/dbms/src/Interpreters/MetricLog.h
@@ -0,0 +1,39 @@
+#pragma once
+#include <Interpreters/SystemLog.h>
+#include <Interpreters/AsynchronousMetrics.h>
+
+namespace DB
+{
+
+using Poco::Message;
+
+struct MetricLogElement
+{
+    time_t event_time{};
+    UInt64 milliseconds{};
+
+    static std::string name() { return "MetricLog"; }
+    static Block createBlock();
+    void appendToBlock(Block & block) const;
+};
+
+class MetricLog : public SystemLog<MetricLogElement>
+{
+    using SystemLog<MetricLogElement>::SystemLog;
+
+public:
+    /// Launches a background thread to collect metrics with interval
+    void startCollectMetric(size_t collect_interval_milliseconds_);
+
+    /// Stop background thread. Call before shutdown.
+    void stopCollectMetric();
+
+private:
+    void metricThreadFunction();
+
+    ThreadFromGlobalPool metric_flush_thread;
+    size_t collect_interval_milliseconds;
+    std::atomic<bool> is_shutdown_metric_thread{false};
+};
+
+}
diff --git a/dbms/src/Interpreters/SystemLog.cpp b/dbms/src/Interpreters/SystemLog.cpp
@@ -4,6 +4,7 @@
 #include <Interpreters/PartLog.h>
 #include <Interpreters/TextLog.h>
 #include <Interpreters/TraceLog.h>
+#include <Interpreters/MetricLog.h>
 
 #include <Poco/Util/AbstractConfiguration.h>
 
@@ -48,6 +49,13 @@ SystemLogs::SystemLogs(Context & global_context, const Poco::Util::AbstractConfi
     part_log = createSystemLog<PartLog>(global_context, "system", "part_log", config, "part_log");
     trace_log = createSystemLog<TraceLog>(global_context, "system", "trace_log", config, "trace_log");
     text_log = createSystemLog<TextLog>(global_context, "system", "text_log", config, "text_log");
+    metric_log = createSystemLog<MetricLog>(global_context, "system", "metric_log", config, "metric_log");
+
+    if (metric_log)
+    {
+        size_t collect_interval_milliseconds = config.getUInt64("metric_log.collect_interval_milliseconds");
+        metric_log->startCollectMetric(collect_interval_milliseconds);
+    }
 
     part_log_database = config.getString("part_log.database", "system");
 }
@@ -70,6 +78,11 @@ void SystemLogs::shutdown()
         trace_log->shutdown();
     if (text_log)
         text_log->shutdown();
+    if (metric_log)
+    {
+        metric_log->stopCollectMetric();
+        metric_log->shutdown();
+    }
 }
 
 }
diff --git a/dbms/src/Interpreters/SystemLog.h b/dbms/src/Interpreters/SystemLog.h
@@ -61,6 +61,7 @@ class QueryThreadLog;
 class PartLog;
 class TextLog;
 class TraceLog;
+class MetricLog;
 
 /// System logs should be destroyed in destructor of the last Context and before tables,
 ///  because SystemLog destruction makes insert query while flushing data into underlying tables
@@ -76,6 +77,7 @@ struct SystemLogs
     std::shared_ptr<PartLog> part_log;                  /// Used to log operations with parts
     std::shared_ptr<TraceLog> trace_log;                /// Used to log traces from query profiler
     std::shared_ptr<TextLog> text_log;                  /// Used to log all text messages.
+    std::shared_ptr<MetricLog> metric_log;              /// Used to log all metrics.
 
     String part_log_database;
 };

diff --git a/dbms/src/Storages/MergeTree/MergeTreeSequentialBlockInputStream.cpp b/dbms/src/Storages/MergeTree/MergeTreeSequentialBlockInputStream.cpp
@@ -27,9 +27,7 @@ MergeTreeSequentialBlockInputStream::MergeTreeSequentialBlockInputStream(
         std::stringstream message;
         message << "Reading " << data_part->getMarksCount() << " marks from part " << data_part->name
             << ", total " << data_part->rows_count
-            << " rows starting from the beginning of the part, columns: ";
-        for (size_t i = 0, size = columns_to_read.size(); i < size; ++i)
-            message << (i == 0 ? "" : ", ") << columns_to_read[i];
+            << " rows starting from the beginning of the part";
 
         LOG_TRACE(log, message.rdbuf());
     }

diff --git a/dbms/tests/queries/0_stateless/00990_metric_log_table_not_empty.reference b/dbms/tests/queries/0_stateless/00990_metric_log_table_not_empty.reference
@@ -0,0 +1 @@
+1
diff --git a/dbms/tests/queries/0_stateless/00990_metric_log_table_not_empty.sql b/dbms/tests/queries/0_stateless/00990_metric_log_table_not_empty.sql
@@ -0,0 +1,5 @@
+select sleep(2) format Null; --More than collect_interval_milliseconds
+
+system flush logs;
+
+select count()>0 from system.metric_log
diff --git a/docker/test/stateful/Dockerfile b/docker/test/stateful/Dockerfile
@@ -20,6 +20,7 @@ CMD dpkg -i package_folder/clickhouse-common-static_*.deb; \
     ln -s /usr/share/clickhouse-test/config/listen.xml /etc/clickhouse-server/config.d/; \
     ln -s /usr/share/clickhouse-test/config/part_log.xml /etc/clickhouse-server/config.d/; \
     ln -s /usr/share/clickhouse-test/config/text_log.xml /etc/clickhouse-server/config.d/; \
+    ln -s /usr/share/clickhouse-test/config/metric_log.xml /etc/clickhouse-server/config.d/; \
     ln -s /usr/share/clickhouse-test/config/log_queries.xml /etc/clickhouse-server/users.d/; \
     ln -s /usr/share/clickhouse-test/config/readonly.xml /etc/clickhouse-server/users.d/; \
     ln -s /usr/share/clickhouse-test/config/ints_dictionary.xml /etc/clickhouse-server/; \

diff --git a/docker/test/stateful_with_coverage/run.sh b/docker/test/stateful_with_coverage/run.sh
@@ -46,6 +46,7 @@ ln -s /usr/share/clickhouse-test/config/zookeeper.xml /etc/clickhouse-server/con
     ln -s /usr/share/clickhouse-test/config/listen.xml /etc/clickhouse-server/config.d/; \
     ln -s /usr/share/clickhouse-test/config/part_log.xml /etc/clickhouse-server/config.d/; \
     ln -s /usr/share/clickhouse-test/config/text_log.xml /etc/clickhouse-server/config.d/; \
+    ln -s /usr/share/clickhouse-test/config/metric_log.xml /etc/clickhouse-server/config.d/; \
     ln -s /usr/share/clickhouse-test/config/log_queries.xml /etc/clickhouse-server/users.d/; \
     ln -s /usr/share/clickhouse-test/config/readonly.xml /etc/clickhouse-server/users.d/; \
     ln -s /usr/share/clickhouse-test/config/ints_dictionary.xml /etc/clickhouse-server/; \

diff --git a/docker/test/stateless/Dockerfile b/docker/test/stateless/Dockerfile
@@ -40,6 +40,7 @@ CMD dpkg -i package_folder/clickhouse-common-static_*.deb; \
     ln -s /usr/share/clickhouse-test/config/listen.xml /etc/clickhouse-server/config.d/; \
     ln -s /usr/share/clickhouse-test/config/part_log.xml /etc/clickhouse-server/config.d/; \
     ln -s /usr/share/clickhouse-test/config/text_log.xml /etc/clickhouse-server/config.d/; \
+    ln -s /usr/share/clickhouse-test/config/metric_log.xml /etc/clickhouse-server/config.d/; \
     ln -s /usr/share/clickhouse-test/config/query_masking_rules.xml /etc/clickhouse-server/config.d/; \
     ln -s /usr/share/clickhouse-test/config/log_queries.xml /etc/clickhouse-server/users.d/; \
     ln -s /usr/share/clickhouse-test/config/readonly.xml /etc/clickhouse-server/users.d/; \

diff --git a/docker/test/stateless_with_coverage/run.sh b/docker/test/stateless_with_coverage/run.sh
@@ -48,6 +48,7 @@ ln -s /usr/share/clickhouse-test/config/zookeeper.xml /etc/clickhouse-server/con
     ln -s /usr/share/clickhouse-test/config/listen.xml /etc/clickhouse-server/config.d/; \
     ln -s /usr/share/clickhouse-test/config/part_log.xml /etc/clickhouse-server/config.d/; \
     ln -s /usr/share/clickhouse-test/config/text_log.xml /etc/clickhouse-server/config.d/; \
+    ln -s /usr/share/clickhouse-test/config/metric_log.xml /etc/clickhouse-server/config.d/; \
     ln -s /usr/share/clickhouse-test/config/log_queries.xml /etc/clickhouse-server/users.d/; \
     ln -s /usr/share/clickhouse-test/config/readonly.xml /etc/clickhouse-server/users.d/; \
     ln -s /usr/share/clickhouse-test/config/ints_dictionary.xml /etc/clickhouse-server/; \