ClickHouse/programs/server/MetricsTransmitter.cpp

Ignoring revisions in .git-blame-ignore-revs. Click here to bypass and see the normal blame view.

135 lines
4.2 KiB
C++
Raw Normal View History

2016-01-17 13:34:36 +00:00
#include "MetricsTransmitter.h"
#include <Common/AsynchronousMetrics.h>
#include <Common/CurrentMetrics.h>
#include <Common/Exception.h>
#include <Common/setThreadName.h>
2022-04-27 15:05:45 +00:00
#include <Daemon/BaseDaemon.h>
#include <Poco/Util/Application.h>
#include <Poco/Util/LayeredConfiguration.h>
2016-01-17 13:34:36 +00:00
namespace DB
{
2019-07-04 19:08:37 +00:00
MetricsTransmitter::MetricsTransmitter(
2019-08-03 11:02:40 +00:00
const Poco::Util::AbstractConfiguration & config, const std::string & config_name_, const AsynchronousMetrics & async_metrics_)
: async_metrics(async_metrics_), config_name(config_name_)
2019-07-04 19:08:37 +00:00
{
interval_seconds = config.getInt(config_name + ".interval", 60);
send_events = config.getBool(config_name + ".events", true);
2019-09-22 22:57:06 +00:00
send_events_cumulative = config.getBool(config_name + ".events_cumulative", false);
2019-07-04 19:08:37 +00:00
send_metrics = config.getBool(config_name + ".metrics", true);
send_asynchronous_metrics = config.getBool(config_name + ".asynchronous_metrics", true);
thread = ThreadFromGlobalPool{&MetricsTransmitter::run, this};
2019-07-04 19:08:37 +00:00
}
2016-01-17 13:34:36 +00:00
MetricsTransmitter::~MetricsTransmitter()
{
try
{
{
2019-01-02 06:44:36 +00:00
std::lock_guard lock{mutex};
2016-01-17 13:34:36 +00:00
quit = true;
}
2016-01-17 13:34:36 +00:00
cond.notify_one();
thread->join();
2016-01-17 13:34:36 +00:00
}
catch (...)
{
2016-10-23 10:52:32 +00:00
DB::tryLogCurrentException(__PRETTY_FUNCTION__);
2016-01-17 13:34:36 +00:00
}
}
void MetricsTransmitter::run()
{
2019-07-04 19:08:37 +00:00
const std::string thread_name = "MetrTx" + std::to_string(interval_seconds);
setThreadName(thread_name.c_str());
2018-01-10 00:04:08 +00:00
const auto get_next_time = [](size_t seconds)
{
/// To avoid time drift and transmit values exactly each interval:
/// next time aligned to system seconds
/// (60s -> every minute at 00 seconds, 5s -> every minute:[00, 05, 15 ... 55]s, 3600 -> every hour:00:00
return std::chrono::system_clock::time_point(
(std::chrono::duration_cast<std::chrono::seconds>(std::chrono::system_clock::now().time_since_epoch()) / seconds) * seconds
+ std::chrono::seconds(seconds));
2016-01-17 13:34:36 +00:00
};
std::vector<ProfileEvents::Count> prev_counters(ProfileEvents::end());
2019-01-02 06:44:36 +00:00
std::unique_lock lock{mutex};
2016-01-17 13:34:36 +00:00
while (true)
{
2019-07-04 19:08:37 +00:00
if (cond.wait_until(lock, get_next_time(interval_seconds), [this] { return quit; }))
2016-01-17 13:34:36 +00:00
break;
transmit(prev_counters);
2016-01-17 13:34:36 +00:00
}
}
2016-11-01 17:56:25 +00:00
void MetricsTransmitter::transmit(std::vector<ProfileEvents::Count> & prev_counters)
2016-01-17 13:34:36 +00:00
{
auto async_metrics_values = async_metrics.getValues();
GraphiteWriter::KeyValueVector<ssize_t> key_vals{};
key_vals.reserve(ProfileEvents::end() + CurrentMetrics::end() + async_metrics_values.size());
2019-07-04 19:08:37 +00:00
if (send_events)
2016-01-17 13:34:36 +00:00
{
for (ProfileEvents::Event i = ProfileEvents::Event(0), end = ProfileEvents::end(); i < end; ++i)
{
const auto counter = ProfileEvents::global_counters[i].load(std::memory_order_relaxed);
const auto counter_increment = counter - prev_counters[i];
prev_counters[i] = counter;
std::string key{ProfileEvents::getName(static_cast<ProfileEvents::Event>(i))};
key_vals.emplace_back(profile_events_path_prefix + key, counter_increment);
}
2016-01-17 13:34:36 +00:00
}
2019-09-22 22:57:06 +00:00
if (send_events_cumulative)
{
for (ProfileEvents::Event i = ProfileEvents::Event(0), end = ProfileEvents::end(); i < end; ++i)
{
const auto counter = ProfileEvents::global_counters[i].load(std::memory_order_relaxed);
std::string key{ProfileEvents::getName(static_cast<ProfileEvents::Event>(i))};
2019-09-22 22:57:06 +00:00
key_vals.emplace_back(profile_events_cumulative_path_prefix + key, counter);
}
}
2019-07-04 19:08:37 +00:00
if (send_metrics)
{
for (CurrentMetrics::Metric i = CurrentMetrics::Metric(0), end = CurrentMetrics::end(); i < end; ++i)
{
const auto value = CurrentMetrics::values[i].load(std::memory_order_relaxed);
std::string key{CurrentMetrics::getName(static_cast<CurrentMetrics::Metric>(i))};
key_vals.emplace_back(current_metrics_path_prefix + key, value);
}
}
2019-07-04 19:08:37 +00:00
if (send_asynchronous_metrics)
{
for (const auto & name_value : async_metrics_values)
{
key_vals.emplace_back(asynchronous_metrics_path_prefix + name_value.first, name_value.second.value);
}
}
2020-03-09 03:38:43 +00:00
if (!key_vals.empty())
BaseDaemon::instance().writeToGraphite(key_vals, config_name);
2016-01-17 13:34:36 +00:00
}
2019-07-04 19:08:37 +00:00
2016-01-17 13:34:36 +00:00
}