Commit 977f00bd authored by Dominik Charousset's avatar Dominik Charousset

Merge branch 'topic/neverlord/prometheus-collector'

parents 7bb918fc 590b5283
...@@ -11,6 +11,11 @@ ...@@ -11,6 +11,11 @@
#include "caf/detail/core_export.hpp" #include "caf/detail/core_export.hpp"
#include "caf/fwd.hpp" #include "caf/fwd.hpp"
#include "caf/string_view.hpp" #include "caf/string_view.hpp"
#include "caf/telemetry/counter.hpp"
#include "caf/telemetry/gauge.hpp"
#include "caf/telemetry/histogram.hpp"
#include "caf/timespan.hpp"
#include "caf/timestamp.hpp"
namespace caf::telemetry::collector { namespace caf::telemetry::collector {
...@@ -29,77 +34,146 @@ public: ...@@ -29,77 +34,146 @@ public:
/// Returns the minimum scrape interval, i.e., the minimum time that needs to /// Returns the minimum scrape interval, i.e., the minimum time that needs to
/// pass before `collect_from` iterates the registry to re-fill the buffer. /// pass before `collect_from` iterates the registry to re-fill the buffer.
time_t min_scrape_interval() const noexcept { [[nodiscard]] timespan min_scrape_interval() const noexcept {
return min_scrape_interval_; return min_scrape_interval_;
} }
/// Sets the minimum scrape interval to `value`. /// Sets the minimum scrape interval to `value`.
void min_scrape_interval(time_t value) noexcept { void min_scrape_interval(timespan value) noexcept {
min_scrape_interval_ = value; min_scrape_interval_ = value;
} }
// -- collect API ------------------------------------------------------------ /// Returns the time point of the last scrape.
[[nodiscard]] timestamp last_scrape() const noexcept {
return last_scrape_;
}
/// Applies this collector to the registry, filling the character buffer while /// Returns a string view into the internal buffer.
/// collecting metrics. /// @warning This view may become invalid when calling any non-const member
/// @param registry Source for the metrics. /// function on the collector object.
/// @param now Timestamp as time since UNIX epoch in seconds. [[nodiscard]] string_view str() const noexcept {
/// @returns a view into the filled buffer. return {buf_.data(), buf_.size()};
string_view collect_from(const metric_registry& registry, time_t now); }
/// Reverts the collector back to its initial state, clearing all buffers.
void reset();
// -- scraping API -----------------------------------------------------------
/// Begins a new scrape if `last_scape() + min_scrape_interval() <= now`.
/// @returns `true` if the collector started a new scrape or `false` to
/// signal that the caller shall use the last result via `str()`
/// since it has not expired yet.
[[nodiscard]] bool begin_scrape(timestamp now = make_timestamp());
/// Cleans up any temporary state before accessing `str()` for obtaining the
/// scrape result.
void end_scrape();
// -- appending into the internal buffer -------------------------------------
template <class T>
void
append_counter(const metric_family* family, const metric* instance, T value) {
append_impl(family, "counter", instance, value);
}
template <class T>
void
append_gauge(const metric_family* family, const metric* instance, T value) {
append_impl(family, "gauge", instance, value);
}
void append_histogram(const metric_family* family, const metric* instance,
span<const int_histogram::bucket_type> buckets,
int64_t sum);
void append_histogram(const metric_family* family, const metric* instance,
span<const dbl_histogram::bucket_type> buckets,
double sum);
// -- collect API ------------------------------------------------------------
/// Applies this collector to the registry, filling the character buffer while /// Applies this collector to the registry, filling the character buffer while
/// collecting metrics. Uses the current system time as timestamp. /// collecting metrics. Automatically calls `begin_scrape` and `end_scrape` as
/// needed.
/// @param registry Source for the metrics. /// @param registry Source for the metrics.
/// @param now Current system time.
/// @returns a view into the filled buffer. /// @returns a view into the filled buffer.
string_view collect_from(const metric_registry& registry); string_view collect_from(const metric_registry& registry,
timestamp now = make_timestamp());
// -- call operators for the metric registry --------------------------------- // -- call operators for the metric registry ---------------------------------
void operator()(const metric_family* family, const metric* instance, void operator()(const metric_family* family, const metric* instance,
const dbl_counter* counter); const dbl_counter* counter) {
append_counter(family, instance, counter->value());
}
void operator()(const metric_family* family, const metric* instance, void operator()(const metric_family* family, const metric* instance,
const int_counter* counter); const int_counter* counter) {
append_counter(family, instance, counter->value());
}
void operator()(const metric_family* family, const metric* instance, void operator()(const metric_family* family, const metric* instance,
const dbl_gauge* gauge); const dbl_gauge* gauge) {
append_gauge(family, instance, gauge->value());
}
void operator()(const metric_family* family, const metric* instance, void operator()(const metric_family* family, const metric* instance,
const int_gauge* gauge); const int_gauge* gauge) {
append_gauge(family, instance, gauge->value());
}
void operator()(const metric_family* family, const metric* instance, void operator()(const metric_family* family, const metric* instance,
const dbl_histogram* val); const dbl_histogram* val) {
append_histogram(family, instance, val->buckets(), val->sum());
}
void operator()(const metric_family* family, const metric* instance, void operator()(const metric_family* family, const metric* instance,
const int_histogram* val); const int_histogram* val) {
append_histogram(family, instance, val->buckets(), val->sum());
}
private: private:
// -- implementation details -------------------------------------------------
/// Sets `current_family_` if not pointing to `family` already. When setting /// Sets `current_family_` if not pointing to `family` already. When setting
/// the member variable, also writes meta information to `buf_`. /// the member variable, also writes meta information to `buf_`.
void set_current_family(const metric_family* family, void set_current_family(const metric_family* family,
string_view prometheus_type); string_view prometheus_type);
template <class ValueType> void append_impl(const metric_family* family, string_view prometheus_type,
void append_histogram(const metric_family* family, const metric* instance, const metric* instance, int64_t value);
const histogram<ValueType>* val);
void append_impl(const metric_family* family, string_view prometheus_type,
const metric* instance, double value);
template <class BucketType, class ValueType>
void append_histogram_impl(const metric_family* family,
const metric* instance,
span<const BucketType> buckets, ValueType sum);
// -- member variables -------------------------------------------------------
/// Stores the generated text output. /// Stores the generated text output.
char_buffer buf_; char_buffer buf_;
/// Current timestamp. /// Current timestamp.
time_t now_ = 0; timestamp last_scrape_ = timestamp{timespan{0}};
/// Caches type information and help text for a metric. /// Caches type information and help text for a metric.
std::unordered_map<const metric_family*, char_buffer> meta_info_; std::unordered_map<const metric_family*, char_buffer> family_info_;
/// Caches type information and help text for a metric. /// Caches variable names for each bucket of a histogram as well as for the
std::unordered_map<const metric*, std::vector<char_buffer>> virtual_metrics_; /// implicit sum and count fields.
std::unordered_map<const metric*, std::vector<char_buffer>> histogram_info_;
/// Caches which metric family is currently collected. /// Caches which metric family is currently collected.
const metric_family* current_family_ = nullptr; const metric_family* current_family_ = nullptr;
/// Minimum time between re-iterating the registry. /// Minimum time between re-iterating the registry.
time_t min_scrape_interval_ = 0; timespan min_scrape_interval_ = timespan{0};
}; };
} // namespace caf::telemetry::collector } // namespace caf::telemetry::collector
...@@ -25,13 +25,14 @@ struct ms_timestamp { ...@@ -25,13 +25,14 @@ struct ms_timestamp {
int64_t value; int64_t value;
/// Converts seconds-since-epoch to milliseconds-since-epoch /// Converts seconds-since-epoch to milliseconds-since-epoch
explicit ms_timestamp(time_t from) : value(from * int64_t{1000}) { explicit ms_timestamp(timestamp from) noexcept {
// nop using ms_dur = std::chrono::duration<int64_t, std::milli>;
value = std::chrono::duration_cast<ms_dur>(from.time_since_epoch()).count();
} }
ms_timestamp(const ms_timestamp&) = default; ms_timestamp(const ms_timestamp&) noexcept = default;
ms_timestamp& operator=(const ms_timestamp&) = default; ms_timestamp& operator=(const ms_timestamp&) noexcept = default;
}; };
// Converts separators such as '.' and '-' to underlines to follow the // Converts separators such as '.' and '-' to underlines to follow the
...@@ -171,67 +172,69 @@ void append(prometheus::char_buffer& buf, const prometheus::char_buffer& x, ...@@ -171,67 +172,69 @@ void append(prometheus::char_buffer& buf, const prometheus::char_buffer& x,
} // namespace } // namespace
string_view prometheus::collect_from(const metric_registry& registry, // -- properties ---------------------------------------------------------------
time_t now) {
if (!buf_.empty() && now - now_ < min_scrape_interval_) void prometheus::reset() {
return {buf_.data(), buf_.size()};
buf_.clear(); buf_.clear();
now_ = now; last_scrape_ = timestamp{timespan{0}};
registry.collect(*this); family_info_.clear();
histogram_info_.clear();
current_family_ = nullptr; current_family_ = nullptr;
return {buf_.data(), buf_.size()}; min_scrape_interval_ = timespan{0};
} }
string_view prometheus::collect_from(const metric_registry& registry) { // -- scraping API -------------------------------------------------------------
return collect_from(registry, time(NULL));
}
void prometheus::operator()(const metric_family* family, const metric* instance, bool prometheus::begin_scrape(timestamp now) {
const dbl_counter* counter) { if (buf_.empty() || last_scrape_ + min_scrape_interval_ <= now) {
set_current_family(family, "counter"); buf_.clear();
append(buf_, family, instance, ' ', counter->value(), ' ', ms_timestamp{now_}, last_scrape_ = now;
'\n'); current_family_ = nullptr;
return true;
} else {
return false;
}
} }
void prometheus::operator()(const metric_family* family, const metric* instance, void prometheus::end_scrape() {
const int_counter* counter) { // nop
set_current_family(family, "counter");
append(buf_, family, instance, ' ', counter->value(), ' ', ms_timestamp{now_},
'\n');
} }
void prometheus::operator()(const metric_family* family, const metric* instance, // -- appending into the internal buffer ---------------------------------------
const dbl_gauge* gauge) {
set_current_family(family, "gauge");
append(buf_, family, instance, ' ', gauge->value(), ' ', ms_timestamp{now_},
'\n');
}
void prometheus::operator()(const metric_family* family, const metric* instance, void prometheus::append_histogram(
const int_gauge* gauge) { const metric_family* family, const metric* instance,
set_current_family(family, "gauge"); span<const int_histogram::bucket_type> buckets, int64_t sum) {
append(buf_, family, instance, ' ', gauge->value(), ' ', ms_timestamp{now_}, append_histogram_impl(family, instance, buckets, sum);
'\n');
} }
void prometheus::operator()(const metric_family* family, const metric* instance, void prometheus::append_histogram(
const dbl_histogram* val) { const metric_family* family, const metric* instance,
append_histogram(family, instance, val); span<const dbl_histogram::bucket_type> buckets, double sum) {
append_histogram_impl(family, instance, buckets, sum);
} }
void prometheus::operator()(const metric_family* family, const metric* instance, // -- collect API --------------------------------------------------------------
const int_histogram* val) {
append_histogram(family, instance, val); string_view prometheus::collect_from(const metric_registry& registry,
timestamp now) {
if (begin_scrape(now)) {
registry.collect(*this);
end_scrape();
}
return str();
} }
// -- implementation details ---------------------------------------------------
void prometheus::set_current_family(const metric_family* family, void prometheus::set_current_family(const metric_family* family,
string_view prometheus_type) { string_view prometheus_type) {
if (current_family_ == family) if (current_family_ == family)
return; return;
current_family_ = family; current_family_ = family;
auto i = meta_info_.find(family); auto i = family_info_.find(family);
if (i == meta_info_.end()) { if (i == family_info_.end()) {
i = meta_info_.emplace(family, char_buffer{}).first; i = family_info_.emplace(family, char_buffer{}).first;
if (!family->helptext().empty()) if (!family->helptext().empty())
append(i->second, "# HELP ", family, ' ', family->helptext(), '\n'); append(i->second, "# HELP ", family, ' ', family->helptext(), '\n');
append(i->second, "# TYPE ", family, ' ', prometheus_type, '\n'); append(i->second, "# TYPE ", family, ' ', prometheus_type, '\n');
...@@ -239,17 +242,32 @@ void prometheus::set_current_family(const metric_family* family, ...@@ -239,17 +242,32 @@ void prometheus::set_current_family(const metric_family* family,
buf_.insert(buf_.end(), i->second.begin(), i->second.end()); buf_.insert(buf_.end(), i->second.begin(), i->second.end());
} }
void prometheus::append_impl(const metric_family* family,
string_view prometheus_type,
const metric* instance, int64_t value) {
set_current_family(family, prometheus_type);
append(buf_, family, instance, ' ', value, ' ', ms_timestamp{last_scrape_},
'\n');
}
void prometheus::append_impl(const metric_family* family,
string_view prometheus_type,
const metric* instance, double value) {
set_current_family(family, prometheus_type);
append(buf_, family, instance, ' ', value, ' ', ms_timestamp{last_scrape_},
'\n');
}
namespace { namespace {
template <class ValueType> template <class BucketType>
auto make_virtual_metrics(const metric_family* family, const metric* instance, auto make_histogram_info(const metric_family* family, const metric* instance,
const histogram<ValueType>* val) { span<const BucketType> buckets) {
std::vector<prometheus::char_buffer> result; std::vector<prometheus::char_buffer> result;
auto add_result = [&](auto&&... xs) { auto add_result = [&](auto&&... xs) {
result.emplace_back(); result.emplace_back();
append(result.back(), std::forward<decltype(xs)>(xs)...); append(result.back(), std::forward<decltype(xs)>(xs)...);
}; };
auto buckets = val->buckets();
auto num_buckets = buckets.size(); auto num_buckets = buckets.size();
CAF_ASSERT(num_buckets > 1); CAF_ASSERT(num_buckets > 1);
auto labels = instance->labels(); auto labels = instance->labels();
...@@ -273,26 +291,26 @@ auto make_virtual_metrics(const metric_family* family, const metric* instance, ...@@ -273,26 +291,26 @@ auto make_virtual_metrics(const metric_family* family, const metric* instance,
} // namespace } // namespace
template <class ValueType> template <class BucketType, class ValueType>
void prometheus::append_histogram(const metric_family* family, void prometheus::append_histogram_impl(const metric_family* family,
const metric* instance, const metric* instance,
const histogram<ValueType>* val) { span<const BucketType> buckets,
auto i = virtual_metrics_.find(instance); ValueType sum) {
if (i == virtual_metrics_.end()) { auto i = histogram_info_.find(instance);
auto metrics = make_virtual_metrics(family, instance, val); if (i == histogram_info_.end()) {
i = virtual_metrics_.emplace(instance, std::move(metrics)).first; auto info = make_histogram_info(family, instance, buckets);
i = histogram_info_.emplace(instance, std::move(info)).first;
} }
set_current_family(family, "histogram"); set_current_family(family, "histogram");
auto& vm = i->second; auto& vm = i->second;
auto buckets = val->buckets();
auto acc = ValueType{0}; auto acc = ValueType{0};
auto index = size_t{0}; auto index = size_t{0};
for (; index < buckets.size(); ++index) { for (; index < buckets.size(); ++index) {
acc += buckets[index].count.value(); acc += buckets[index].count.value();
append(buf_, vm[index], acc, ' ', ms_timestamp{now_}, '\n'); append(buf_, vm[index], acc, ' ', ms_timestamp{last_scrape_}, '\n');
} }
append(buf_, vm[index++], val->sum(), ' ', ms_timestamp{now_}, '\n'); append(buf_, vm[index++], sum, ' ', ms_timestamp{last_scrape_}, '\n');
append(buf_, vm[index++], acc, ' ', ms_timestamp{now_}, '\n'); append(buf_, vm[index++], acc, ' ', ms_timestamp{last_scrape_}, '\n');
} }
} // namespace caf::telemetry::collector } // namespace caf::telemetry::collector
...@@ -15,6 +15,8 @@ using namespace caf; ...@@ -15,6 +15,8 @@ using namespace caf;
using namespace caf::literals; using namespace caf::literals;
using namespace caf::telemetry; using namespace caf::telemetry;
using namespace std::literals;
namespace { namespace {
struct fixture { struct fixture {
...@@ -44,7 +46,7 @@ CAF_TEST(the Prometheus collector generates text output) { ...@@ -44,7 +46,7 @@ CAF_TEST(the Prometheus collector generates text output) {
h->observe(3); h->observe(3);
h->observe(4); h->observe(4);
h->observe(7); h->observe(7);
CAF_CHECK_EQUAL(exporter.collect_from(registry, 42), CAF_CHECK_EQUAL(exporter.collect_from(registry, timestamp{42s}),
R"(# HELP foo_bar_seconds Some value without labels. R"(# HELP foo_bar_seconds Some value without labels.
# TYPE foo_bar_seconds gauge # TYPE foo_bar_seconds gauge
foo_bar_seconds 123 42000 foo_bar_seconds 123 42000
......
Markdown is supported
0%
or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment