Commit cfc0baa5 authored by Dominik Charousset's avatar Dominik Charousset

Integrate telemetry for number of running actors

parent f1c30206
...@@ -33,6 +33,7 @@ ...@@ -33,6 +33,7 @@
#include "caf/detail/core_export.hpp" #include "caf/detail/core_export.hpp"
#include "caf/detail/shared_spinlock.hpp" #include "caf/detail/shared_spinlock.hpp"
#include "caf/fwd.hpp" #include "caf/fwd.hpp"
#include "caf/telemetry/int_gauge.hpp"
namespace caf { namespace caf {
...@@ -121,7 +122,7 @@ private: ...@@ -121,7 +122,7 @@ private:
actor_registry(actor_system& sys); actor_registry(actor_system& sys);
std::atomic<size_t> running_; telemetry::int_gauge* running_;
mutable std::mutex running_mtx_; mutable std::mutex running_mtx_;
mutable std::condition_variable running_cv_; mutable std::condition_variable running_cv_;
......
...@@ -49,6 +49,7 @@ ...@@ -49,6 +49,7 @@
#include "caf/scoped_execution_unit.hpp" #include "caf/scoped_execution_unit.hpp"
#include "caf/spawn_options.hpp" #include "caf/spawn_options.hpp"
#include "caf/string_algorithms.hpp" #include "caf/string_algorithms.hpp"
#include "caf/telemetry/metric_registry.hpp"
#include "caf/type_id.hpp" #include "caf/type_id.hpp"
namespace caf::detail { namespace caf::detail {
...@@ -229,8 +230,20 @@ public: ...@@ -229,8 +230,20 @@ public:
return assignable(xs, message_types<T>()); return assignable(xs, message_types<T>());
} }
/// Returns the telemetry registry for this system.
telemetry::metric_registry& telemetry() noexcept {
return telemetry_;
}
/// Returns the telemetry registry for this system.
const telemetry::metric_registry& telemetry() const noexcept {
return telemetry_;
}
/// Returns the host-local identifier for this system. /// Returns the host-local identifier for this system.
const node_id& node() const; const node_id& node() const {
return node_;
}
/// Returns the scheduler instance. /// Returns the scheduler instance.
scheduler::abstract_coordinator& scheduler(); scheduler::abstract_coordinator& scheduler();
...@@ -586,6 +599,9 @@ private: ...@@ -586,6 +599,9 @@ private:
// -- member variables ------------------------------------------------------- // -- member variables -------------------------------------------------------
/// Manages all metrics collected by the system.
telemetry::metric_registry telemetry_;
/// Provides system-wide callbacks for several actor operations. /// Provides system-wide callbacks for several actor operations.
actor_profiler* profiler_; actor_profiler* profiler_;
......
...@@ -67,6 +67,18 @@ public: ...@@ -67,6 +67,18 @@ public:
return value_.load(); return value_.load();
} }
/// Increments the gauge by 1.
/// @returns The new value of the gauge.
int64_t operator++() noexcept {
return ++value_;
}
/// Decrements the gauge by 1.
/// @returns The new value of the gauge.
int64_t operator--() noexcept {
return --value_;
}
private: private:
std::atomic<int64_t> value_; std::atomic<int64_t> value_;
}; };
......
...@@ -45,6 +45,14 @@ public: ...@@ -45,6 +45,14 @@ public:
return impl_; return impl_;
} }
Type* impl_ptr() noexcept {
return &impl_;
}
const Type* impl_ptr() const noexcept {
return &impl_;
}
private: private:
Type impl_; Type impl_;
}; };
......
...@@ -32,6 +32,12 @@ namespace caf::telemetry { ...@@ -32,6 +32,12 @@ namespace caf::telemetry {
/// Manages a collection of metric families. /// Manages a collection of metric families.
class CAF_CORE_EXPORT metric_registry { class CAF_CORE_EXPORT metric_registry {
public: public:
template <class Type>
using metric_family_ptr = std::unique_ptr<metric_family_impl<Type>>;
template <class Type>
using metric_family_container = std::vector<metric_family_ptr<Type>>;
metric_registry(); metric_registry();
~metric_registry(); ~metric_registry();
...@@ -55,24 +61,87 @@ public: ...@@ -55,24 +61,87 @@ public:
std::vector<std::string> label_names, std::string helptext, std::vector<std::string> label_names, std::string helptext,
std::string unit = "1", bool is_sum = false); std::string unit = "1", bool is_sum = false);
/// Adds a new metric family to the registry.
/// @param prefix The prefix (namespace) this family belongs to. Usually the
/// application or protocol name, e.g., `http`. The prefix `caf`
/// as well as prefixes starting with an underscore are
/// reserved.
/// @param name The human-readable name of the metric, e.g., `requests`.
/// @param label_names Names for all label dimensions of the metric.
/// @param helptext Short explanation of the metric.
/// @param unit Unit of measurement. Please use base units such as `bytes` or
/// `seconds` (prefer lowercase). The pseudo-unit `1` identifies
/// dimensionless counts.
/// @param is_sum Setting this to `true` indicates that this metric adds
/// something up to a total, where only the total value is of
/// interest. For example, the total number of HTTP requests.
template <class Type>
metric_family_impl<Type>*
add_family(std::string prefix, std::string name,
std::vector<std::string> label_names, std::string helptext,
std::string unit = "1", bool is_sum = false) {
using family_type = metric_family_impl<Type>;
std::sort(label_names.begin(), label_names.end());
auto ptr = std::make_unique<family_type>(std::move(prefix), std::move(name),
std::move(label_names),
std::move(helptext),
std::move(unit), is_sum);
std::unique_lock<std::mutex> guard{families_mx_};
assert_unregistered(ptr->prefix(), ptr->name());
auto& families = container_by_type<Type>();
families.emplace_back(std::move(ptr));
return families.back().get();
}
/// Adds a new metric singleton, i.e., a family without label dimensions and
/// thus exactly one metric instance.
/// @param prefix The prefix (namespace) this family belongs to. Usually the
/// application or protocol name, e.g., `http`. The prefix `caf`
/// as well as prefixes starting with an underscore are
/// reserved.
/// @param name The human-readable name of the metric, e.g., `requests`.
/// @param helptext Short explanation of the metric.
/// @param unit Unit of measurement. Please use base units such as `bytes` or
/// `seconds` (prefer lowercase). The pseudo-unit `1` identifies
/// dimensionless counts.
/// @param is_sum Setting this to `true` indicates that this metric adds
/// something up to a total, where only the total value is of
/// interest. For example, the total number of HTTP requests.
template <class Type>
Type* add_singleton(std::string prefix, std::string name,
std::string helptext, std::string unit = "1",
bool is_sum = false) {
auto fptr = add_family<Type>(std::move(prefix), std::move(name), {},
std::move(helptext), std::move(unit), is_sum);
return fptr->get_or_add({});
}
telemetry::int_gauge* int_gauge(string_view prefix, string_view name, telemetry::int_gauge* int_gauge(string_view prefix, string_view name,
std::vector<label_view> labels); std::vector<label_view> labels);
template <class Collector> template <class Collector>
void collect(Collector& collector) const { void collect(Collector& collector) const {
for (auto& ptr : int_gauges) std::unique_lock<std::mutex> guard{families_mx_};
for (auto& ptr : int_gauges_)
ptr->collect(collector); ptr->collect(collector);
} }
private: private:
void add_int_gauge_family(std::string prefix, std::string name, /// @pre `families_mx_` is locked.
std::vector<std::string> label_names, void assert_unregistered(const std::string& prefix, const std::string& name);
std::string helptext);
template <class Type> template <class Type>
using metric_family_ptr = std::unique_ptr<metric_family_impl<Type>>; metric_family_container<Type>& container_by_type();
std::vector<metric_family_ptr<telemetry::int_gauge>> int_gauges; mutable std::mutex families_mx_;
metric_family_container<telemetry::int_gauge> int_gauges_;
}; };
template <>
inline metric_registry::metric_family_container<int_gauge>&
metric_registry::container_by_type<int_gauge>() {
return int_gauges_;
}
} // namespace caf::telemetry } // namespace caf::telemetry
...@@ -48,8 +48,9 @@ actor_registry::~actor_registry() { ...@@ -48,8 +48,9 @@ actor_registry::~actor_registry() {
// nop // nop
} }
actor_registry::actor_registry(actor_system& sys) : running_(0), system_(sys) { actor_registry::actor_registry(actor_system& sys) : system_(sys) {
// nop running_ = sys.telemetry().add_singleton<telemetry::int_gauge>(
"caf", "running_actors", "Number of currently running actors.");
} }
strong_actor_ptr actor_registry::get_impl(actor_id key) const { strong_actor_ptr actor_registry::get_impl(actor_id key) const {
...@@ -96,19 +97,19 @@ void actor_registry::erase(actor_id key) { ...@@ -96,19 +97,19 @@ void actor_registry::erase(actor_id key) {
void actor_registry::inc_running() { void actor_registry::inc_running() {
# if CAF_LOG_LEVEL >= CAF_LOG_LEVEL_DEBUG # if CAF_LOG_LEVEL >= CAF_LOG_LEVEL_DEBUG
auto value = ++running_; auto value = ++*running_;
CAF_LOG_DEBUG(CAF_ARG(value)); CAF_LOG_DEBUG(CAF_ARG(value));
# else # else
++running_; ++*running_;
# endif # endif
} }
size_t actor_registry::running() const { size_t actor_registry::running() const {
return running_.load(); return running_->value();
} }
void actor_registry::dec_running() { void actor_registry::dec_running() {
size_t new_val = --running_; size_t new_val = --*running_;
if (new_val <= 1) { if (new_val <= 1) {
std::unique_lock<std::mutex> guard(running_mtx_); std::unique_lock<std::mutex> guard(running_mtx_);
running_cv_.notify_all(); running_cv_.notify_all();
...@@ -120,8 +121,8 @@ void actor_registry::await_running_count_equal(size_t expected) const { ...@@ -120,8 +121,8 @@ void actor_registry::await_running_count_equal(size_t expected) const {
CAF_ASSERT(expected == 0 || expected == 1); CAF_ASSERT(expected == 0 || expected == 1);
CAF_LOG_TRACE(CAF_ARG(expected)); CAF_LOG_TRACE(CAF_ARG(expected));
std::unique_lock<std::mutex> guard{running_mtx_}; std::unique_lock<std::mutex> guard{running_mtx_};
while (running_ != expected) { while (running_->value() != static_cast<int64_t>(expected)) {
CAF_LOG_DEBUG(CAF_ARG(running_.load())); CAF_LOG_DEBUG(CAF_ARG(running_->value()));
running_cv_.wait(guard); running_cv_.wait(guard);
} }
} }
......
...@@ -344,11 +344,6 @@ actor_system::~actor_system() { ...@@ -344,11 +344,6 @@ actor_system::~actor_system() {
logger_dtor_cv_.wait(guard); logger_dtor_cv_.wait(guard);
} }
/// Returns the host-local identifier for this system.
const node_id& actor_system::node() const {
return node_;
}
/// Returns the scheduler instance. /// Returns the scheduler instance.
scheduler::abstract_coordinator& actor_system::scheduler() { scheduler::abstract_coordinator& actor_system::scheduler() {
using ptr = scheduler::abstract_coordinator*; using ptr = scheduler::abstract_coordinator*;
......
...@@ -19,6 +19,7 @@ ...@@ -19,6 +19,7 @@
#include "caf/telemetry/metric_registry.hpp" #include "caf/telemetry/metric_registry.hpp"
#include "caf/config.hpp" #include "caf/config.hpp"
#include "caf/raise_error.hpp"
#include "caf/telemetry/int_gauge.hpp" #include "caf/telemetry/int_gauge.hpp"
#include "caf/telemetry/metric_family_impl.hpp" #include "caf/telemetry/metric_family_impl.hpp"
#include "caf/telemetry/metric_impl.hpp" #include "caf/telemetry/metric_impl.hpp"
...@@ -52,40 +53,56 @@ metric_registry::int_gauge(string_view prefix, string_view name, ...@@ -52,40 +53,56 @@ metric_registry::int_gauge(string_view prefix, string_view name,
family->label_names().begin(), family->label_names().begin(),
family->label_names().end(), lbl_cmp); family->label_names().end(), lbl_cmp);
}; };
auto i = std::find_if(int_gauges.begin(), int_gauges.end(), matches); using family_type = metric_family_impl<telemetry::int_gauge>;
if (i == int_gauges.end()) { family_type* fptr;
return nullptr; {
std::unique_lock<std::mutex> guard{families_mx_};
auto i = std::find_if(int_gauges_.begin(), int_gauges_.end(), matches);
if (i == int_gauges_.end()) {
std::string prefix_str{prefix.begin(), prefix.end()};
std::string name_str{name.begin(), name.end()};
assert_unregistered(prefix_str, name_str);
std::vector<std::string> label_names;
label_names.reserve(labels.size());
for (auto& lbl : labels) {
auto lbl_name = lbl.name();
label_names.emplace_back(std::string{lbl_name.begin(), lbl_name.end()});
} }
return (*i)->get_or_add(labels); auto ptr = std::make_unique<family_type>(std::move(prefix_str),
} std::move(name_str),
std::move(label_names), "", "1",
namespace { false);
i = int_gauges_.emplace(i, std::move(ptr));
template <class Type> }
using metric_family_ptr = std::unique_ptr<metric_family_impl<Type>>; fptr = i->get();
}
template <class T, class... Ts> return fptr->get_or_add(labels);
void emplace_family(std::vector<metric_family_ptr<T>>& families, Ts&&... xs) {
using family_type = metric_family_impl<T>;
families.emplace_back(std::make_unique<family_type>(std::forward<Ts>(xs)...));
} }
} // namespace
void metric_registry::add_family(metric_type type, std::string prefix, void metric_registry::add_family(metric_type type, std::string prefix,
std::string name, std::string name,
std::vector<std::string> label_names, std::vector<std::string> label_names,
std::string helptext, std::string unit, std::string helptext, std::string unit,
bool is_sum) { bool is_sum) {
std::sort(label_names.begin(), label_names.end()); namespace t = caf::telemetry;
switch (type) { switch (type) {
case metric_type::int_gauge: case metric_type::int_gauge:
return emplace_family(int_gauges, std::move(prefix), std::move(name), add_family<t::int_gauge>(std::move(prefix), std::move(name),
std::move(label_names), std::move(helptext), std::move(label_names), std::move(helptext),
std::move(unit), is_sum); std::move(unit), is_sum);
default:
break; break;
default:
CAF_RAISE_ERROR("invalid metric type");
} }
} }
void metric_registry::assert_unregistered(const std::string& prefix,
const std::string& name) {
auto matches = [&](const auto& ptr) {
return ptr->prefix() == prefix && ptr->name() == name;
};
if (std::any_of(int_gauges_.begin(), int_gauges_.end(), matches))
CAF_RAISE_ERROR("prefix and name already belong to an int gauge family");
}
} // namespace caf::telemetry } // namespace caf::telemetry
...@@ -116,14 +116,14 @@ CAF_TEST(registries allow users to collect all registered metrics) { ...@@ -116,14 +116,14 @@ CAF_TEST(registries allow users to collect all registered metrics) {
"seconds", true); "seconds", true);
registry.add_family(metric_type::int_gauge, "caf", "running_actors", {"node"}, registry.add_family(metric_type::int_gauge, "caf", "running_actors", {"node"},
"How many actors are currently running?"); "How many actors are currently running?");
registry.add_family(metric_type::int_gauge, "caf", "mailbox_size", {"name"},
"How many message are currently in mailboxes?");
registry.int_gauge("foo", "bar", {})->value(123); registry.int_gauge("foo", "bar", {})->value(123);
registry.int_gauge("some", "value", {{"a", "1"}, {"b", "2"}})->value(12); registry.int_gauge("some", "value", {{"a", "1"}, {"b", "2"}})->value(12);
registry.int_gauge("some", "value", {{"b", "1"}, {"a", "2"}})->value(21); registry.int_gauge("some", "value", {{"b", "1"}, {"a", "2"}})->value(21);
registry.int_gauge("other", "value", {{"x", "true"}})->value(31337); registry.int_gauge("other", "value", {{"x", "true"}})->value(31337);
auto running_actors = bind("caf", "running_actors"); auto running_actors = bind("caf", "running_actors");
running_actors.int_gauge({{"node", "localhost"}})->value(42); running_actors.int_gauge({{"node", "localhost"}})->value(42);
// Note: not registering caf_mailbox_size previously forces CAF to create it
// lazily when adding the first gauge.
auto mailbox_size = bind("caf", "mailbox_size"); auto mailbox_size = bind("caf", "mailbox_size");
mailbox_size.int_gauge({{"name", "printer"}})->value(3); mailbox_size.int_gauge({{"name", "printer"}})->value(3);
mailbox_size.int_gauge({{"name", "parser"}})->value(12); mailbox_size.int_gauge({{"name", "parser"}})->value(12);
......
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