Commit 590b5283 authored by Dominik Charousset's avatar Dominik Charousset

Refactor API of the Prometheus collector

parent 7bb918fc
......@@ -11,6 +11,11 @@
#include "caf/detail/core_export.hpp"
#include "caf/fwd.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 {
......@@ -29,77 +34,146 @@ public:
/// 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.
time_t min_scrape_interval() const noexcept {
[[nodiscard]] timespan min_scrape_interval() const noexcept {
return min_scrape_interval_;
}
/// 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;
}
// -- 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
/// collecting metrics.
/// @param registry Source for the metrics.
/// @param now Timestamp as time since UNIX epoch in seconds.
/// @returns a view into the filled buffer.
string_view collect_from(const metric_registry& registry, time_t now);
/// Returns a string view into the internal buffer.
/// @warning This view may become invalid when calling any non-const member
/// function on the collector object.
[[nodiscard]] string_view str() const noexcept {
return {buf_.data(), buf_.size()};
}
/// 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
/// 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 now Current system time.
/// @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 ---------------------------------
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,
const int_counter* counter);
const int_counter* counter) {
append_counter(family, instance, counter->value());
}
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,
const int_gauge* gauge);
const int_gauge* gauge) {
append_gauge(family, instance, gauge->value());
}
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,
const int_histogram* val);
const int_histogram* val) {
append_histogram(family, instance, val->buckets(), val->sum());
}
private:
// -- implementation details -------------------------------------------------
/// Sets `current_family_` if not pointing to `family` already. When setting
/// the member variable, also writes meta information to `buf_`.
void set_current_family(const metric_family* family,
string_view prometheus_type);
template <class ValueType>
void append_histogram(const metric_family* family, const metric* instance,
const histogram<ValueType>* val);
void append_impl(const metric_family* family, string_view prometheus_type,
const metric* instance, int64_t value);
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.
char_buffer buf_;
/// Current timestamp.
time_t now_ = 0;
timestamp last_scrape_ = timestamp{timespan{0}};
/// 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.
std::unordered_map<const metric*, std::vector<char_buffer>> virtual_metrics_;
/// Caches variable names for each bucket of a histogram as well as for the
/// implicit sum and count fields.
std::unordered_map<const metric*, std::vector<char_buffer>> histogram_info_;
/// Caches which metric family is currently collected.
const metric_family* current_family_ = nullptr;
/// Minimum time between re-iterating the registry.
time_t min_scrape_interval_ = 0;
timespan min_scrape_interval_ = timespan{0};
};
} // namespace caf::telemetry::collector
......@@ -25,13 +25,14 @@ struct ms_timestamp {
int64_t value;
/// Converts seconds-since-epoch to milliseconds-since-epoch
explicit ms_timestamp(time_t from) : value(from * int64_t{1000}) {
// nop
explicit ms_timestamp(timestamp from) noexcept {
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
......@@ -171,67 +172,69 @@ void append(prometheus::char_buffer& buf, const prometheus::char_buffer& x,
} // namespace
string_view prometheus::collect_from(const metric_registry& registry,
time_t now) {
if (!buf_.empty() && now - now_ < min_scrape_interval_)
return {buf_.data(), buf_.size()};
// -- properties ---------------------------------------------------------------
void prometheus::reset() {
buf_.clear();
now_ = now;
registry.collect(*this);
last_scrape_ = timestamp{timespan{0}};
family_info_.clear();
histogram_info_.clear();
current_family_ = nullptr;
return {buf_.data(), buf_.size()};
min_scrape_interval_ = timespan{0};
}
string_view prometheus::collect_from(const metric_registry& registry) {
return collect_from(registry, time(NULL));
}
// -- scraping API -------------------------------------------------------------
void prometheus::operator()(const metric_family* family, const metric* instance,
const dbl_counter* counter) {
set_current_family(family, "counter");
append(buf_, family, instance, ' ', counter->value(), ' ', ms_timestamp{now_},
'\n');
bool prometheus::begin_scrape(timestamp now) {
if (buf_.empty() || last_scrape_ + min_scrape_interval_ <= now) {
buf_.clear();
last_scrape_ = now;
current_family_ = nullptr;
return true;
} else {
return false;
}
}
void prometheus::operator()(const metric_family* family, const metric* instance,
const int_counter* counter) {
set_current_family(family, "counter");
append(buf_, family, instance, ' ', counter->value(), ' ', ms_timestamp{now_},
'\n');
void prometheus::end_scrape() {
// nop
}
void prometheus::operator()(const metric_family* family, const metric* instance,
const dbl_gauge* gauge) {
set_current_family(family, "gauge");
append(buf_, family, instance, ' ', gauge->value(), ' ', ms_timestamp{now_},
'\n');
}
// -- appending into the internal buffer ---------------------------------------
void prometheus::operator()(const metric_family* family, const metric* instance,
const int_gauge* gauge) {
set_current_family(family, "gauge");
append(buf_, family, instance, ' ', gauge->value(), ' ', ms_timestamp{now_},
'\n');
void prometheus::append_histogram(
const metric_family* family, const metric* instance,
span<const int_histogram::bucket_type> buckets, int64_t sum) {
append_histogram_impl(family, instance, buckets, sum);
}
void prometheus::operator()(const metric_family* family, const metric* instance,
const dbl_histogram* val) {
append_histogram(family, instance, val);
void prometheus::append_histogram(
const metric_family* family, const metric* instance,
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,
const int_histogram* val) {
append_histogram(family, instance, val);
// -- collect API --------------------------------------------------------------
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,
string_view prometheus_type) {
if (current_family_ == family)
return;
current_family_ = family;
auto i = meta_info_.find(family);
if (i == meta_info_.end()) {
i = meta_info_.emplace(family, char_buffer{}).first;
auto i = family_info_.find(family);
if (i == family_info_.end()) {
i = family_info_.emplace(family, char_buffer{}).first;
if (!family->helptext().empty())
append(i->second, "# HELP ", family, ' ', family->helptext(), '\n');
append(i->second, "# TYPE ", family, ' ', prometheus_type, '\n');
......@@ -239,17 +242,32 @@ void prometheus::set_current_family(const metric_family* family,
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 {
template <class ValueType>
auto make_virtual_metrics(const metric_family* family, const metric* instance,
const histogram<ValueType>* val) {
template <class BucketType>
auto make_histogram_info(const metric_family* family, const metric* instance,
span<const BucketType> buckets) {
std::vector<prometheus::char_buffer> result;
auto add_result = [&](auto&&... xs) {
result.emplace_back();
append(result.back(), std::forward<decltype(xs)>(xs)...);
};
auto buckets = val->buckets();
auto num_buckets = buckets.size();
CAF_ASSERT(num_buckets > 1);
auto labels = instance->labels();
......@@ -273,26 +291,26 @@ auto make_virtual_metrics(const metric_family* family, const metric* instance,
} // namespace
template <class ValueType>
void prometheus::append_histogram(const metric_family* family,
template <class BucketType, class ValueType>
void prometheus::append_histogram_impl(const metric_family* family,
const metric* instance,
const histogram<ValueType>* val) {
auto i = virtual_metrics_.find(instance);
if (i == virtual_metrics_.end()) {
auto metrics = make_virtual_metrics(family, instance, val);
i = virtual_metrics_.emplace(instance, std::move(metrics)).first;
span<const BucketType> buckets,
ValueType sum) {
auto i = histogram_info_.find(instance);
if (i == histogram_info_.end()) {
auto info = make_histogram_info(family, instance, buckets);
i = histogram_info_.emplace(instance, std::move(info)).first;
}
set_current_family(family, "histogram");
auto& vm = i->second;
auto buckets = val->buckets();
auto acc = ValueType{0};
auto index = size_t{0};
for (; index < buckets.size(); ++index) {
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++], acc, ' ', ms_timestamp{now_}, '\n');
append(buf_, vm[index++], sum, ' ', ms_timestamp{last_scrape_}, '\n');
append(buf_, vm[index++], acc, ' ', ms_timestamp{last_scrape_}, '\n');
}
} // namespace caf::telemetry::collector
......@@ -15,6 +15,8 @@ using namespace caf;
using namespace caf::literals;
using namespace caf::telemetry;
using namespace std::literals;
namespace {
struct fixture {
......@@ -44,7 +46,7 @@ CAF_TEST(the Prometheus collector generates text output) {
h->observe(3);
h->observe(4);
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.
# TYPE foo_bar_seconds gauge
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