Commit 13b5d670 authored by Dominik Charousset's avatar Dominik Charousset

Merge branch 'topic/telemetry'

parents d8035474 9b40c088
......@@ -44,6 +44,10 @@ is based on [Keep a Changelog](https://keepachangelog.com).
RFC4122-compliant `uuid` class.
- The new trait class `is_error_code_enum` allows users to enable conversion of
custom error code enums to `error` and `error_code`.
- CAF now enables users to tap into internal CAF metrics as well as adding their
own instrumentation! Since this addition is too large to cover in a changelog
entry, please have a look at the new *Metrics* Section of the manual to learn
more.
### Deprecated
......@@ -123,6 +127,12 @@ is based on [Keep a Changelog](https://keepachangelog.com).
- All parsing functions in `actor_system_config` that take an input stream
exclusively use the new configuration syntax (please consult the manual for
details and examples for the configuration syntax).
- The returned string of `name()` must not change during the lifetime of an
actor. Hence, `stateful_actor` now only considers static `name` members in its
`State` for overriding this function. CAF always assumed names belonging to
*types*, but did not enforce it because the name was only used for logging.
Since the new metrics use this name for filtering now, we enforce static names
in order to help avoid hard-to-find issues with the filtering mechanism.
### Removed
......
......@@ -83,6 +83,7 @@ add_library(libcaf_core_obj OBJECT ${CAF_CORE_HEADERS}
src/detail/get_mac_addresses.cpp
src/detail/get_process_id.cpp
src/detail/get_root_uuid.cpp
src/detail/glob_match.cpp
src/detail/ini_consumer.cpp
src/detail/invoke_result_visitor.cpp
src/detail/message_builder_element.cpp
......@@ -163,6 +164,12 @@ add_library(libcaf_core_obj OBJECT ${CAF_CORE_HEADERS}
src/stream_priority_strings.cpp
src/string_algorithms.cpp
src/string_view.cpp
src/telemetry/collector/prometheus.cpp
src/telemetry/label.cpp
src/telemetry/label_view.cpp
src/telemetry/metric.cpp
src/telemetry/metric_family.cpp
src/telemetry/metric_registry.cpp
src/term.cpp
src/test_credit_controller.cpp
src/thread_hook.cpp
......@@ -305,6 +312,13 @@ caf_add_test_suites(caf-core-test
string_algorithms
string_view
sum_type
telemetry.collector.prometheus
telemetry.counter
telemetry.gauge
telemetry.histogram
telemetry.label
telemetry.metric_registry
telemetry.timer
thread_hook
tracing_data
type_id_list
......
......@@ -133,6 +133,7 @@ public:
static constexpr int is_initialized_flag = 0x0010; // event-based actors
static constexpr int is_blocking_flag = 0x0020; // blocking_actor
static constexpr int is_detached_flag = 0x0040; // local_actor
static constexpr int collects_metrics_flag = 0x0080; // local_actor
static constexpr int is_serializable_flag = 0x0100; // local_actor
static constexpr int is_migrated_from_flag = 0x0200; // local_actor
static constexpr int has_used_aout_flag = 0x0400; // local_actor
......
......@@ -105,6 +105,10 @@ public:
void on_destroy() override;
void setup_metrics() {
// nop
}
protected:
void on_cleanup(const error& reason) override;
......
......@@ -38,6 +38,10 @@ public:
/// Invokes cleanup code.
virtual void kill_proxy(execution_unit* ctx, error reason) = 0;
void setup_metrics() {
// nop
}
};
} // namespace caf
......@@ -33,6 +33,7 @@
#include "caf/detail/core_export.hpp"
#include "caf/detail/shared_spinlock.hpp"
#include "caf/fwd.hpp"
#include "caf/telemetry/int_gauge.hpp"
namespace caf {
......@@ -121,7 +122,6 @@ private:
actor_registry(actor_system& sys);
std::atomic<size_t> running_;
mutable std::mutex running_mtx_;
mutable std::condition_variable running_cv_;
......
......@@ -49,6 +49,7 @@
#include "caf/scoped_execution_unit.hpp"
#include "caf/spawn_options.hpp"
#include "caf/string_algorithms.hpp"
#include "caf/telemetry/metric_registry.hpp"
#include "caf/type_id.hpp"
namespace caf::detail {
......@@ -172,6 +173,39 @@ public:
virtual void demonitor(const node_id& node, const actor_addr& observer) = 0;
};
/// Metrics that the actor system collects by default.
/// @warning Do not modify these metrics in user code. Some may be used by the
/// system for synchronization.
struct base_metrics_t {
/// Counts the number of messages that where rejected because the target
/// mailbox was closed or did not exist.
telemetry::int_counter* rejected_messages;
/// Counts the total number of processed messages.
telemetry::int_counter* processed_messages;
/// Tracks the current number of running actors in the system.
telemetry::int_gauge* running_actors;
/// Counts the total number of messages that wait in a mailbox.
telemetry::int_gauge* queued_messages;
};
/// Metrics that some actors may collect in addition to the base metrics. All
/// families in this set use the label dimension *name* (the user-defined name
/// of the actor).
struct actor_metric_families_t {
/// Samples how long the actor needs to process messages.
telemetry::dbl_histogram_family* processing_time_family = nullptr;
/// Samples how long a message waits in the mailbox before the actor
/// processes it.
telemetry::dbl_histogram_family* mailbox_time_family = nullptr;
/// Counts how many messages are currently waiting in the mailbox.
telemetry::int_gauge_family* mailbox_size_family = nullptr;
};
/// @warning The system stores a reference to `cfg`, which means the
/// config object must outlive the actor system.
explicit actor_system(actor_system_config& cfg);
......@@ -229,8 +263,20 @@ public:
return assignable(xs, message_types<T>());
}
/// Returns the metrics registry for this system.
telemetry::metric_registry& metrics() noexcept {
return metrics_;
}
/// Returns the metrics registry for this system.
const telemetry::metric_registry& metrics() const noexcept {
return metrics_;
}
/// Returns the host-local identifier for this system.
const node_id& node() const;
const node_id& node() const {
return node_;
}
/// Returns the scheduler instance.
scheduler::abstract_coordinator& scheduler();
......@@ -390,8 +436,8 @@ public:
/// range `[first, last)`.
/// @private
template <spawn_options Os, class Iter, class F, class... Ts>
infer_handle_from_fun_t<F>
spawn_fun_in_groups(actor_config& cfg, Iter first, Iter second, F& fun,
infer_handle_from_fun_t<F> spawn_fun_in_groups(actor_config& cfg, Iter first,
Iter second, F& fun,
Ts&&... xs) {
using impl = infer_impl_from_fun_t<F>;
check_invariants<impl>();
......@@ -499,6 +545,14 @@ public:
/// @warning must be called by thread which is about to terminate
void thread_terminates();
const auto& metrics_actors_includes() const noexcept {
return metrics_actors_includes_;
}
const auto& metrics_actors_excludes() const noexcept {
return metrics_actors_excludes_;
}
template <class C, spawn_options Os, class... Ts>
infer_handle_from_class_t<C> spawn_impl(actor_config& cfg, Ts&&... xs) {
static_assert(is_unbound(Os),
......@@ -543,8 +597,8 @@ public:
profiler_->after_processing(self, result);
}
void
profiler_before_sending(const local_actor& self, mailbox_element& element) {
void profiler_before_sending(const local_actor& self,
mailbox_element& element) {
if (profiler_)
profiler_->before_sending(self, element);
}
......@@ -556,6 +610,18 @@ public:
profiler_->before_sending_scheduled(self, timeout, element);
}
base_metrics_t& base_metrics() noexcept {
return base_metrics_;
}
const auto& base_metrics() const noexcept {
return base_metrics_;
}
const auto& actor_metric_families() const noexcept {
return actor_metric_families_;
}
tracing_data_factory* tracing_context() const noexcept {
return tracing_context_;
}
......@@ -592,6 +658,12 @@ private:
/// Used to generate ascending actor IDs.
std::atomic<size_t> ids_;
/// Manages all metrics collected by the system.
telemetry::metric_registry metrics_;
/// Stores all metrics that the actor system collects by default.
base_metrics_t base_metrics_;
/// Identifies this actor system in a distributed setting.
node_id node_;
......@@ -643,6 +715,17 @@ private:
/// Stores the system-wide factory for deserializing tracing data.
tracing_data_factory* tracing_context_;
/// Caches the configuration parameter `caf.metrics-filters.actors.includes`
/// for faster lookups at runtime.
std::vector<std::string> metrics_actors_includes_;
/// Caches the configuration parameter `caf.metrics-filters.actors.excludes`
/// for faster lookups at runtime.
std::vector<std::string> metrics_actors_excludes_;
/// Caches families for optional actor metrics.
actor_metric_families_t actor_metric_families_;
};
} // namespace caf
......@@ -46,6 +46,10 @@ public:
message_types_set message_types() const override;
void setup_metrics() {
// nop
}
protected:
void on_cleanup(const error& reason) override;
......
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright 2011-2020 Dominik Charousset *
* *
* Distributed under the terms and conditions of the BSD 3-Clause License or *
* (at your option) under the terms and conditions of the Boost Software *
* License 1.0. See accompanying files LICENSE and LICENSE_ALTERNATIVE. *
* *
* If you did not receive a copy of the license files, see *
* http://opensource.org/licenses/BSD-3-Clause and *
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#pragma once
namespace caf::detail {
// gitignore-style pathname globbing.
bool glob_match(const char* str, const char* glob);
} // namespace caf::detail
......@@ -35,13 +35,14 @@ template <class Self, class SelfHandle, class Handle, class... Ts>
void profiled_send(Self* self, SelfHandle&& src, const Handle& dst,
message_id msg_id, std::vector<strong_actor_ptr> stages,
execution_unit* context, Ts&&... xs) {
CAF_IGNORE_UNUSED(self);
if (dst) {
auto element = make_mailbox_element(std::forward<SelfHandle>(src), msg_id,
std::move(stages),
std::forward<Ts>(xs)...);
CAF_BEFORE_SENDING(self, *element);
dst->enqueue(std::move(element), context);
} else {
self->home_system().base_metrics().rejected_messages->inc();
}
}
......@@ -49,7 +50,6 @@ template <class Self, class SelfHandle, class Handle, class... Ts>
void profiled_send(Self* self, SelfHandle&& src, const Handle& dst,
actor_clock& clock, actor_clock::time_point timeout,
message_id msg_id, Ts&&... xs) {
CAF_IGNORE_UNUSED(self);
if (dst) {
if constexpr (std::is_same<Handle, group>::value) {
clock.schedule_message(timeout, dst, std::forward<SelfHandle>(src),
......@@ -61,6 +61,8 @@ void profiled_send(Self* self, SelfHandle&& src, const Handle& dst,
clock.schedule_message(timeout, actor_cast<strong_actor_ptr>(dst),
std::move(element));
}
} else {
self->home_system().base_metrics().rejected_messages->inc();
}
}
......
......@@ -229,6 +229,72 @@ struct subscriber_base;
} // namespace mixin
// -- telemetry API ------------------------------------------------------------
namespace telemetry {
class component;
class dbl_gauge;
class int_gauge;
class label;
class label_view;
class metric;
class metric_family;
class metric_registry;
class timer;
enum class metric_type : uint8_t;
template <class ValueType>
class counter;
template <class ValueType>
class histogram;
template <class Type>
class metric_family_impl;
template <class Type>
class metric_impl;
using dbl_counter = counter<double>;
using dbl_histogram = histogram<double>;
using int_counter = counter<int64_t>;
using int_histogram = histogram<int64_t>;
using dbl_counter_family = metric_family_impl<dbl_counter>;
using dbl_histogram_family = metric_family_impl<dbl_histogram>;
using dbl_gauge_family = metric_family_impl<dbl_gauge>;
using int_counter_family = metric_family_impl<int_counter>;
using int_histogram_family = metric_family_impl<int_histogram>;
using int_gauge_family = metric_family_impl<int_gauge>;
} // namespace telemetry
namespace detail {
template <class>
struct gauge_oracle;
template <>
struct gauge_oracle<double> {
using type = telemetry::dbl_gauge;
};
template <>
struct gauge_oracle<int64_t> {
using type = telemetry::int_gauge;
};
} // namespace detail
namespace telemetry {
template <class ValueType>
using gauge = typename detail::gauge_oracle<ValueType>::type;
} // namespace telemetry
// -- I/O classes --------------------------------------------------------------
namespace io {
......
......@@ -170,7 +170,7 @@ public:
/// @returns `true` if `f` consumed at least one item.
template <class F>
bool consume(F& f) noexcept(noexcept(f(std::declval<value_type&>()))) {
return new_round(0, f).consumed_items;
return new_round(0, f).consumed_items > 0;
}
/// Run a new round with `quantum`, dispatching all tasks to `consumer`.
......@@ -178,12 +178,12 @@ public:
new_round_result new_round(deficit_type quantum, F& consumer) noexcept(
noexcept(consumer(std::declval<value_type&>()))) {
if (list_.empty())
return {false, false};
return {0, false};
deficit_ += quantum;
long consumed = 0;
auto ptr = next();
if (ptr == nullptr)
return {false, false};
return {0, false};
size_t consumed = 0;
do {
auto consumer_res = consumer(*ptr);
switch (consumer_res) {
......@@ -194,7 +194,7 @@ public:
cache_.push_back(ptr.release());
if (list_.empty()) {
deficit_ = 0;
return {consumed != 0, false};
return {consumed, false};
}
break;
case task_result::resume:
......@@ -202,7 +202,7 @@ public:
flush_cache();
if (list_.empty()) {
deficit_ = 0;
return {consumed != 0, false};
return {consumed, false};
}
break;
default:
......@@ -210,11 +210,11 @@ public:
flush_cache();
if (list_.empty())
deficit_ = 0;
return {consumed != 0, consumer_res == task_result::stop_all};
return {consumed, consumer_res == task_result::stop_all};
}
ptr = next();
} while (ptr != nullptr);
return {consumed != 0, false};
return {consumed, false};
}
cache_type& cache() noexcept {
......
......@@ -95,25 +95,27 @@ public:
/// @returns `true` if at least one item was consumed, `false` otherwise.
template <class F>
new_round_result new_round(deficit_type quantum, F& consumer) {
size_t consumed = 0;
if (!super::empty()) {
deficit_ += quantum;
auto ptr = next();
if (ptr == nullptr)
return {false, false};
return {0, false};
do {
++consumed;
switch (consumer(*ptr)) {
default:
break;
case task_result::stop:
return {true, false};
return {consumed, false};
case task_result::stop_all:
return {true, true};
return {consumed, true};
}
ptr = next();
} while (ptr != nullptr);
return {true, false};
return {consumed, false};
}
return {false, false};
return {consumed, false};
}
private:
......
......@@ -22,15 +22,16 @@
#include <type_traits>
#include "caf/fwd.hpp"
#include "caf/meta/type_name.hpp"
namespace caf::intrusive {
/// Returns the state of a consumer from `new_round`.
struct new_round_result {
/// Denotes whether the consumer accepted at least one element.
bool consumed_items : 1;
size_t consumed_items;
/// Denotes whether the consumer returned `task_result::stop_all`.
bool stop_all : 1;
bool stop_all;
};
constexpr bool operator==(new_round_result x, new_round_result y) {
......@@ -41,13 +42,9 @@ constexpr bool operator!=(new_round_result x, new_round_result y) {
return !(x == y);
}
constexpr new_round_result
make_new_round_result(bool consumed_items, bool stop_all = false) {
return {consumed_items, stop_all};
}
constexpr new_round_result operator|(new_round_result x, new_round_result y) {
return {x.consumed_items || y.consumed_items, x.stop_all || y.stop_all};
template <class Inspector>
typename Inspector::result_type inspect(Inspector& f, new_round_result& x) {
return f(meta::type_name("new_round_result"), x.consumed_items, x.stop_all);
}
} // namespace caf::intrusive
......@@ -110,7 +110,7 @@ public:
/// @returns `true` if at least one item was consumed, `false` otherwise.
template <class F>
new_round_result new_round(deficit_type quantum, F& f) {
bool result = false;
size_t consumed = 0;
bool stopped = false;
for (auto& kvp : qs_) {
if (policy_.enabled(kvp.second)) {
......@@ -118,7 +118,7 @@ public:
if (!stopped) {
new_round_helper<F> g{kvp.first, q, f};
auto res = q.new_round(policy_.quantum(q, quantum), g);
result = res.consumed_items;
consumed += res.consumed_items;
if (res.stop_all)
stopped = true;
} else {
......@@ -129,7 +129,7 @@ public:
}
}
cleanup();
return {result, stopped};
return {consumed, stopped};
}
/// Erases all keys previously marked via `erase_later`.
......
......@@ -186,7 +186,7 @@ private:
template <size_t I, class F>
detail::enable_if_t<I == num_queues, new_round_result>
new_round_recursion(deficit_type, F&) noexcept {
return {false, false};
return {0, false};
}
template <size_t I, class F>
......@@ -202,7 +202,8 @@ private:
inc_deficit_recursion<I + 1>(quantum);
return res;
}
return res | new_round_recursion<I + 1>(quantum, f);
auto sub = new_round_recursion<I + 1>(quantum, f);
return {res.consumed_items + sub.consumed_items, sub.stop_all};
}
template <size_t I>
......
......@@ -49,6 +49,7 @@
#include "caf/response_type.hpp"
#include "caf/resumable.hpp"
#include "caf/spawn_options.hpp"
#include "caf/telemetry/histogram.hpp"
#include "caf/timespan.hpp"
#include "caf/typed_actor.hpp"
#include "caf/typed_response_promise.hpp"
......@@ -64,6 +65,18 @@ public:
/// Defines a monotonic clock suitable for measuring intervals.
using clock_type = std::chrono::steady_clock;
/// Optional metrics collected by individual actors when configured to do so.
struct metrics_t {
/// Samples how long the actor needs to process messages.
telemetry::dbl_histogram* processing_time = nullptr;
/// Samples how long messages wait in the mailbox before being processed.
telemetry::dbl_histogram* mailbox_time = nullptr;
/// Counts how many messages are currently waiting in the mailbox.
telemetry::int_gauge* mailbox_size = nullptr;
};
// -- constructors, destructors, and assignment operators --------------------
local_actor(actor_config& cfg);
......@@ -72,6 +85,11 @@ public:
void on_destroy() override;
// This function performs additional steps to initialize actor-specific
// metrics. It calls virtual functions and thus cannot run as part of the
// constructor.
void setup_metrics();
// -- pure virtual modifiers -------------------------------------------------
virtual void launch(execution_unit* eu, bool lazy, bool hide) = 0;
......@@ -360,6 +378,10 @@ public:
/// @cond PRIVATE
auto& builtin_metrics() noexcept {
return metrics_;
}
template <class ActorHandle>
ActorHandle eval_opts(spawn_options opts, ActorHandle res) {
if (has_monitor_flag(opts))
......@@ -408,6 +430,8 @@ protected:
/// Factory function for returning initial behavior in function-based actors.
detail::unique_function<behavior(local_actor*)> initial_behavior_fac_;
metrics_t metrics_;
};
} // namespace caf
......@@ -18,6 +18,7 @@
#pragma once
#include <chrono>
#include <cstddef>
#include <memory>
......@@ -57,6 +58,27 @@ public:
/// Stores the payload.
message payload;
/// Stores a timestamp for when this element got enqueued.
std::chrono::steady_clock::time_point enqueue_time;
/// Sets `enqueue_time` to the current time.
void set_enqueue_time() {
enqueue_time = std::chrono::steady_clock::now();
}
/// Returns the time between enqueueing the message and `t`.
double seconds_until(std::chrono::steady_clock::time_point t) const {
namespace ch = std::chrono;
using fractional_seconds = ch::duration<double>;
return ch::duration_cast<fractional_seconds>(t - enqueue_time).count();
}
/// Returns the time since calling `set_enqueue_time` in seconds.
double seconds_since_enqueue() const {
namespace ch = std::chrono;
return seconds_until(std::chrono::steady_clock::now());
}
mailbox_element() = default;
mailbox_element(strong_actor_ptr sender, message_id mid,
......
......@@ -44,12 +44,14 @@ R make_actor(actor_id aid, node_id nid, actor_system* sys, Ts&&... xs) {
std::forward<Ts>(xs)...);
}
CAF_LOG_SPAWN_EVENT(ptr->data, args);
ptr->data.setup_metrics();
return {&(ptr->ctrl), false};
}
#endif
CAF_PUSH_AID(aid);
auto ptr = new actor_storage<T>(aid, std::move(nid), sys,
std::forward<Ts>(xs)...);
ptr->data.setup_metrics();
return {&(ptr->ctrl), false};
}
......
......@@ -75,6 +75,7 @@ public:
} else {
self->eq_impl(req_id.response_id(), self->ctrl(), self->context(),
make_error(sec::invalid_argument));
self->home_system().base_metrics().rejected_messages->inc();
}
using response_type
= response_type_t<typename Handle::signatures,
......
......@@ -68,11 +68,13 @@ public:
"statically typed actors can only send() to other "
"statically typed actors; use anon_send() when "
"communicating to groups");
auto self = dptr();
// TODO: consider whether it's feasible to track messages to groups
if (dest) {
auto self = dptr();
dest->eq_impl(make_message_id(P), self->ctrl(), self->context(),
std::forward<Ts>(xs)...);
} else {
self->home_system().base_metrics().rejected_messages->inc();
}
}
......@@ -137,11 +139,13 @@ public:
static_assert(!statically_typed<Subtype>(),
"statically typed actors are not allowed to send to groups");
// TODO: consider whether it's feasible to track messages to groups
if (dest) {
auto self = dptr();
if (dest) {
auto& clock = self->system().clock();
clock.schedule_message(timeout, dest, self->ctrl(),
make_message(std::forward<Ts>(xs)...));
} else {
self->home_system().base_metrics().rejected_messages->inc();
}
}
......
......@@ -49,18 +49,21 @@ public:
static constexpr bool writes_state = false;
template <class... Ts>
[[nodiscard]] auto operator()(Ts&&... xs) {
using result_type = typename Subtype::result_type;
if constexpr (std::is_same<result_type, void>::value) {
template <class Sub = Subtype, class... Ts>
std::enable_if_t<std::is_same<typename Sub::result_type, void>::value>
operator()(Ts&&... xs) {
auto dummy = unit;
static_cast<void>((try_apply(dummy, xs) && ...));
} else {
typename Subtype::result_type result;
}
template <class Sub = Subtype, class... Ts>
std::enable_if_t<!std::is_same<typename Sub::result_type, void>::value,
typename Sub::result_type>
operator()(Ts&&... xs) {
auto result = typename Sub::result_type{};
static_cast<void>((try_apply(result, xs) && ...));
return result;
}
}
private:
template <class Tuple, size_t... Is>
......
......@@ -60,6 +60,7 @@
#include "caf/scheduled_actor.hpp"
#include "caf/sec.hpp"
#include "caf/stream_manager.hpp"
#include "caf/telemetry/timer.hpp"
#include "caf/to_string.hpp"
namespace caf {
......@@ -197,6 +198,7 @@ public:
scheduled_actor* self;
size_t& handled_msgs;
size_t max_throughput;
bool collect_metrics;
/// Consumes upstream messages.
intrusive::task_result operator()(size_t, upstream_queue&,
......@@ -217,6 +219,24 @@ public:
// Consumes asynchronous messages.
intrusive::task_result operator()(mailbox_element& x);
template <class F>
intrusive::task_result run(mailbox_element& x, F body) {
if (collect_metrics) {
auto t0 = std::chrono::steady_clock::now();
auto mbox_time = x.seconds_until(t0);
auto res = body();
if (res != intrusive::task_result::skip) {
auto& builtins = self->builtin_metrics();
telemetry::timer::observe(builtins.processing_time, t0);
builtins.mailbox_time->observe(mbox_time);
builtins.mailbox_size->dec();
}
return res;
} else {
return body();
}
}
};
// -- static helper functions ------------------------------------------------
......
......@@ -79,14 +79,16 @@ public:
const char* name() const override {
if constexpr (detail::has_name<State>::value) {
if constexpr (std::is_convertible<decltype(state.name),
const char*>::value)
return state.name;
else
return state.name.c_str();
if constexpr (!std::is_member_pointer<decltype(&State::name)>::value) {
if constexpr (std::is_convertible<decltype(State::name),
const char*>::value) {
return State::name;
}
} else {
return Base::name();
non_static_name_member(state.name);
}
}
return Base::name();
}
union {
......@@ -95,6 +97,14 @@ public:
/// its reference count drops to zero.
State state;
};
template <class T>
[[deprecated("non-static 'State::name' members have no effect since 0.18")]]
// This function only exists to raise a deprecated warning.
static void
non_static_name_member(const T&) {
// nop
}
};
} // namespace caf
......
......@@ -294,6 +294,14 @@ inline std::string to_string(string_view x) {
} // namespace caf
namespace caf::literals {
constexpr string_view operator""_sv(const char* cstr, size_t len) {
return {cstr, len};
}
} // namespace caf::literals
namespace std {
CAF_CORE_EXPORT std::ostream& operator<<(std::ostream& out, caf::string_view);
......
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright 2011-2020 Dominik Charousset *
* *
* Distributed under the terms and conditions of the BSD 3-Clause License or *
* (at your option) under the terms and conditions of the Boost Software *
* License 1.0. See accompanying files LICENSE and LICENSE_ALTERNATIVE. *
* *
* If you did not receive a copy of the license files, see *
* http://opensource.org/licenses/BSD-3-Clause and *
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#pragma once
#include <ctime>
#include <unordered_map>
#include <vector>
#include "caf/detail/core_export.hpp"
#include "caf/fwd.hpp"
#include "caf/string_view.hpp"
namespace caf::telemetry::collector {
/// Collects system metrics and exports them to the text-based Prometheus
/// format. For a documentation of the format, see: https://git.io/fjgDD.
class CAF_CORE_EXPORT prometheus {
public:
// -- member types -----------------------------------------------------------
/// A buffer for storing UTF-8 characters. Using a vector instead of a
/// `std::string` has slight performance benefits, since the vector does not
/// have to maintain a null-terminator.
using char_buffer = std::vector<char>;
// -- properties -------------------------------------------------------------
/// 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 {
return min_scrape_interval_;
}
/// Sets the minimum scrape interval to `value`.
void min_scrape_interval(time_t value) noexcept {
min_scrape_interval_ = value;
}
// -- collect API ------------------------------------------------------------
/// 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);
/// Applies this collector to the registry, filling the character buffer while
/// collecting metrics. Uses the current system time as timestamp.
/// @param registry Source for the metrics.
/// @returns a view into the filled buffer.
string_view collect_from(const metric_registry& registry);
// -- call operators for the metric registry ---------------------------------
void operator()(const metric_family* family, const metric* instance,
const dbl_counter* counter);
void operator()(const metric_family* family, const metric* instance,
const int_counter* counter);
void operator()(const metric_family* family, const metric* instance,
const dbl_gauge* gauge);
void operator()(const metric_family* family, const metric* instance,
const int_gauge* gauge);
void operator()(const metric_family* family, const metric* instance,
const dbl_histogram* val);
void operator()(const metric_family* family, const metric* instance,
const int_histogram* val);
private:
/// 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);
/// Stores the generated text output.
char_buffer buf_;
/// Current timestamp.
time_t now_ = 0;
/// Caches type information and help text for a metric.
std::unordered_map<const metric_family*, char_buffer> meta_info_;
/// Caches type information and help text for a metric.
std::unordered_map<const metric*, std::vector<char_buffer>> virtual_metrics_;
/// 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;
};
} // namespace caf::telemetry::collector
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright 2011-2020 Dominik Charousset *
* *
* Distributed under the terms and conditions of the BSD 3-Clause License or *
* (at your option) under the terms and conditions of the Boost Software *
* License 1.0. See accompanying files LICENSE and LICENSE_ALTERNATIVE. *
* *
* If you did not receive a copy of the license files, see *
* http://opensource.org/licenses/BSD-3-Clause and *
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#pragma once
#include <type_traits>
#include "caf/config.hpp"
#include "caf/fwd.hpp"
#include "caf/span.hpp"
#include "caf/telemetry/gauge.hpp"
#include "caf/telemetry/label.hpp"
namespace caf::telemetry {
/// A metric that represents a single value that can only go up.
template <class ValueType>
class counter {
public:
// -- member types -----------------------------------------------------------
using value_type = ValueType;
using family_setting = unit_t;
// -- constants --------------------------------------------------------------
static constexpr metric_type runtime_type
= std::is_same<value_type, double>::value ? metric_type::dbl_counter
: metric_type::int_counter;
// -- constructors, destructors, and assignment operators --------------------
counter() noexcept = default;
explicit counter(value_type initial_value) noexcept : gauge_(initial_value) {
// nop
}
explicit counter(span<const label>) noexcept {
// nop
}
// -- modifiers --------------------------------------------------------------
/// Increments the counter by 1.
void inc() noexcept {
gauge_.inc();
}
/// Increments the counter by `amount`.
/// @pre `amount > 0`
void inc(value_type amount) noexcept {
CAF_ASSERT(amount > 0);
gauge_.inc(amount);
}
/// Increments the counter by 1.
/// @returns The new value of the counter.
template <class T = ValueType>
std::enable_if_t<std::is_same<T, int64_t>::value, T> operator++() noexcept {
return ++gauge_;
}
// -- observers --------------------------------------------------------------
/// Returns the current value of the counter.
value_type value() const noexcept {
return gauge_.value();
}
private:
gauge<value_type> gauge_;
};
/// Convenience alias for a counter with value type `double`.
using dbl_counter = counter<double>;
/// Convenience alias for a counter with value type `int64_t`.
using int_counter = counter<int64_t>;
} // namespace caf::telemetry
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright 2011-2020 Dominik Charousset *
* *
* Distributed under the terms and conditions of the BSD 3-Clause License or *
* (at your option) under the terms and conditions of the Boost Software *
* License 1.0. See accompanying files LICENSE and LICENSE_ALTERNATIVE. *
* *
* If you did not receive a copy of the license files, see *
* http://opensource.org/licenses/BSD-3-Clause and *
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#pragma once
#include "caf/detail/core_export.hpp"
#include <atomic>
#include <cstdint>
#include "caf/telemetry/label.hpp"
#include "caf/telemetry/metric_type.hpp"
namespace caf::telemetry {
/// A metric that represents a single integer value that can arbitrarily go up
/// and down.
class CAF_CORE_EXPORT dbl_gauge {
public:
// -- member types -----------------------------------------------------------
using value_type = double;
using family_setting = unit_t;
// -- constants --------------------------------------------------------------
static constexpr metric_type runtime_type = metric_type::dbl_gauge;
// -- constructors, destructors, and assignment operators --------------------
dbl_gauge() noexcept : value_(0) {
// nop
}
explicit dbl_gauge(double value) noexcept : value_(value) {
// nop
}
explicit dbl_gauge(span<const label>) noexcept : value_(0) {
// nop
}
// -- modifiers --------------------------------------------------------------
/// Increments the gauge by 1.
void inc() noexcept {
inc(1.0);
}
/// Increments the gauge by `amount`.
void inc(double amount) noexcept {
auto val = value_.load();
auto new_val = val + amount;
while (!value_.compare_exchange_weak(val, new_val)) {
new_val = val + amount;
}
}
/// Decrements the gauge by 1.
void dec() noexcept {
dec(1.0);
}
/// Decrements the gauge by `amount`.
void dec(double amount) noexcept {
inc(-amount);
}
/// Sets the gauge to `x`.
void value(double x) noexcept {
value_.store(x);
}
// -- observers --------------------------------------------------------------
/// Returns the current value of the gauge.
double value() const noexcept {
return value_.load();
}
private:
std::atomic<double> value_;
};
} // namespace caf::telemetry
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright 2011-2020 Dominik Charousset *
* *
* Distributed under the terms and conditions of the BSD 3-Clause License or *
* (at your option) under the terms and conditions of the Boost Software *
* License 1.0. See accompanying files LICENSE and LICENSE_ALTERNATIVE. *
* *
* If you did not receive a copy of the license files, see *
* http://opensource.org/licenses/BSD-3-Clause and *
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#pragma once
// convenience header for including all gauge types
#include "caf/fwd.hpp"
#include "caf/telemetry/dbl_gauge.hpp"
#include "caf/telemetry/int_gauge.hpp"
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright 2011-2020 Dominik Charousset *
* *
* Distributed under the terms and conditions of the BSD 3-Clause License or *
* (at your option) under the terms and conditions of the Boost Software *
* License 1.0. See accompanying files LICENSE and LICENSE_ALTERNATIVE. *
* *
* If you did not receive a copy of the license files, see *
* http://opensource.org/licenses/BSD-3-Clause and *
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#pragma once
#include <algorithm>
#include <type_traits>
#include "caf/config.hpp"
#include "caf/fwd.hpp"
#include "caf/settings.hpp"
#include "caf/span.hpp"
#include "caf/telemetry/counter.hpp"
#include "caf/telemetry/gauge.hpp"
#include "caf/telemetry/label.hpp"
#include "caf/telemetry/metric_type.hpp"
namespace caf::telemetry {
/// Represent aggregatable distributions of events.
template <class ValueType>
class histogram {
public:
// -- member types -----------------------------------------------------------
using value_type = ValueType;
using gauge_type = gauge<value_type>;
using counter_type = counter<value_type>;
using family_setting = std::vector<value_type>;
struct bucket_type {
value_type upper_bound;
counter_type count;
};
// -- constants --------------------------------------------------------------
static constexpr metric_type runtime_type
= std::is_same<value_type, double>::value ? metric_type::dbl_histogram
: metric_type::int_histogram;
// -- constructors, destructors, and assignment operators --------------------
histogram(span<const label> labels, const settings* cfg,
span<const value_type> upper_bounds) {
if (!init_buckets_from_config(labels, cfg))
init_buckets(upper_bounds);
}
explicit histogram(std::initializer_list<value_type> upper_bounds)
: histogram({}, nullptr,
make_span(upper_bounds.begin(), upper_bounds.size())) {
// nop
}
histogram(const histogram&) = delete;
histogram& operator=(const histogram&) = delete;
~histogram() {
delete[] buckets_;
}
// -- modifiers --------------------------------------------------------------
/// Increments the bucket where the observed value falls into and increments
/// the sum of all observed values.
void observe(value_type value) {
// The last bucket has an upper bound of +inf or int_max, so we'll always
// find a bucket and increment the counters.
for (size_t index = 0;; ++index) {
auto& [upper_bound, count] = buckets_[index];
if (value <= upper_bound) {
count.inc();
sum_.inc(value);
return;
}
}
}
// -- observers --------------------------------------------------------------
/// Returns the ``counter`` objects with the configured upper bounds.
span<const bucket_type> buckets() const noexcept {
return {buckets_, num_buckets_};
}
/// Returns the sum of all observed values.
value_type sum() const noexcept {
return sum_.value();
}
private:
void init_buckets(span<const value_type> upper_bounds) {
CAF_ASSERT(std::is_sorted(upper_bounds.begin(), upper_bounds.end()));
using limits = std::numeric_limits<value_type>;
num_buckets_ = upper_bounds.size() + 1;
buckets_ = new bucket_type[num_buckets_];
size_t index = 0;
for (; index < upper_bounds.size(); ++index)
buckets_[index].upper_bound = upper_bounds[index];
if constexpr (limits::has_infinity)
buckets_[index].upper_bound = limits::infinity();
else
buckets_[index].upper_bound = limits::max();
}
bool init_buckets_from_config(span<const label> labels, const settings* cfg) {
if (cfg == nullptr || labels.empty())
return false;
for (const auto& lbl : labels) {
if (auto ptr = get_if<settings>(cfg, lbl.str())) {
if (auto bounds = get_if<std::vector<value_type>>(ptr, "buckets")) {
std::sort(bounds->begin(), bounds->end());
bounds->erase(std::unique(bounds->begin(), bounds->end()),
bounds->end());
if (bounds->empty())
return false;
init_buckets(*bounds);
return true;
}
}
}
return false;
}
size_t num_buckets_;
bucket_type* buckets_;
gauge_type sum_;
};
/// Convenience alias for a histogram with value type `double`.
using dbl_histogram = histogram<double>;
/// Convenience alias for a histogram with value type `int64_t`.
using int_histogram = histogram<int64_t>;
} // namespace caf::telemetry
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright 2011-2020 Dominik Charousset *
* *
* Distributed under the terms and conditions of the BSD 3-Clause License or *
* (at your option) under the terms and conditions of the Boost Software *
* License 1.0. See accompanying files LICENSE and LICENSE_ALTERNATIVE. *
* *
* If you did not receive a copy of the license files, see *
* http://opensource.org/licenses/BSD-3-Clause and *
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#pragma once
#include "caf/detail/core_export.hpp"
#include <atomic>
#include <cstdint>
#include "caf/fwd.hpp"
#include "caf/span.hpp"
#include "caf/telemetry/label.hpp"
#include "caf/telemetry/metric_type.hpp"
namespace caf::telemetry {
/// A metric that represents a single integer value that can arbitrarily go up
/// and down.
class CAF_CORE_EXPORT int_gauge {
public:
// -- member types -----------------------------------------------------------
using value_type = int64_t;
using family_setting = unit_t;
// -- constants --------------------------------------------------------------
static constexpr metric_type runtime_type = metric_type::int_gauge;
// -- constructors, destructors, and assignment operators --------------------
int_gauge() noexcept : value_(0) {
// nop
}
explicit int_gauge(int64_t value) noexcept : value_(value) {
// nop
}
explicit int_gauge(span<const label>) noexcept : value_(0) {
// nop
}
// -- modifiers --------------------------------------------------------------
/// Increments the gauge by 1.
void inc() noexcept {
++value_;
}
/// Increments the gauge by `amount`.
void inc(int64_t amount) noexcept {
value_.fetch_add(amount);
}
/// Decrements the gauge by 1.
void dec() noexcept {
--value_;
}
/// Decrements the gauge by `amount`.
void dec(int64_t amount) noexcept {
value_.fetch_sub(amount);
}
/// Sets the gauge to `x`.
void value(int64_t x) noexcept {
value_.store(x);
}
/// 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_;
}
// -- observers --------------------------------------------------------------
/// Returns the current value of the gauge.
int64_t value() const noexcept {
return value_.load();
}
private:
std::atomic<int64_t> value_;
};
} // namespace caf::telemetry
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright 2011-2020 Dominik Charousset *
* *
* Distributed under the terms and conditions of the BSD 3-Clause License or *
* (at your option) under the terms and conditions of the Boost Software *
* License 1.0. See accompanying files LICENSE and LICENSE_ALTERNATIVE. *
* *
* If you did not receive a copy of the license files, see *
* http://opensource.org/licenses/BSD-3-Clause and *
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#pragma once
#include <string>
#include "caf/detail/comparable.hpp"
#include "caf/detail/core_export.hpp"
#include "caf/fwd.hpp"
#include "caf/hash/fnv.hpp"
#include "caf/string_view.hpp"
namespace caf::telemetry {
/// An (immutable) key-value pair for adding extra dimensions to metrics.
class CAF_CORE_EXPORT label : detail::comparable<label> {
public:
// -- constructors, destructors, and assignment operators --------------------
label() = delete;
label(label&&) = default;
label(const label&) = default;
label& operator=(label&&) = default;
label& operator=(const label&) = default;
/// @pre `name` matches the regex `[a-zA-Z_:][a-zA-Z0-9_:]*`
label(string_view name, string_view value);
explicit label(const label_view& view);
// -- properties -------------------------------------------------------------
string_view name() const noexcept {
return string_view{str_.data(), name_length_};
}
string_view value() const noexcept {
return string_view{str_.data() + name_length_ + 1,
str_.size() - name_length_ - 1};
}
void value(string_view new_value);
/// Returns the label in `name=value` notation.
const std::string& str() const noexcept {
return str_;
}
// -- comparison -------------------------------------------------------------
int compare(const label& x) const noexcept;
private:
size_t name_length_;
std::string str_;
};
/// Returns the @ref label in `name=value` notation.
/// @relates label
CAF_CORE_EXPORT std::string to_string(const label& x);
} // namespace caf::telemetry
namespace std {
template <>
struct hash<caf::telemetry::label> {
size_t operator()(const caf::telemetry::label& x) const noexcept {
return caf::hash::fnv<size_t>::compute(x.str());
}
};
} // namespace std
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright 2011-2020 Dominik Charousset *
* *
* Distributed under the terms and conditions of the BSD 3-Clause License or *
* (at your option) under the terms and conditions of the Boost Software *
* License 1.0. See accompanying files LICENSE and LICENSE_ALTERNATIVE. *
* *
* If you did not receive a copy of the license files, see *
* http://opensource.org/licenses/BSD-3-Clause and *
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#pragma once
#include "caf/detail/comparable.hpp"
#include "caf/detail/core_export.hpp"
#include "caf/fwd.hpp"
#include "caf/hash/fnv.hpp"
#include "caf/string_view.hpp"
namespace caf::telemetry {
class CAF_CORE_EXPORT label_view : detail::comparable<label_view>,
detail::comparable<label_view, label> {
public:
// -- constructors, destructors, and assignment operators --------------------
label_view() = delete;
label_view(const label_view&) = default;
label_view& operator=(const label_view&) = default;
/// @pre `key` matches the regex `[a-zA-Z_:][a-zA-Z0-9_:]*`
label_view(string_view name, string_view value) : name_(name), value_(value) {
// nop
}
// -- properties -------------------------------------------------------------
string_view name() const noexcept {
return name_;
}
string_view value() const noexcept {
return value_;
}
// -- comparison -------------------------------------------------------------
int compare(const label& x) const noexcept;
int compare(const label_view& x) const noexcept;
private:
string_view name_;
string_view value_;
};
/// Returns the @ref label_view in `name=value` notation.
/// @relates label_view
CAF_CORE_EXPORT std::string to_string(const label_view& x);
} // namespace caf::telemetry
namespace std {
template <>
struct hash<caf::telemetry::label_view> {
size_t operator()(const caf::telemetry::label_view& x) const noexcept {
return caf::hash::fnv<size_t>::compute(x.name(), '=', x.value());
}
};
} // namespace std
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright 2011-2020 Dominik Charousset *
* *
* Distributed under the terms and conditions of the BSD 3-Clause License or *
* (at your option) under the terms and conditions of the Boost Software *
* License 1.0. See accompanying files LICENSE and LICENSE_ALTERNATIVE. *
* *
* If you did not receive a copy of the license files, see *
* http://opensource.org/licenses/BSD-3-Clause and *
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#pragma once
#include <string>
#include <utility>
#include <vector>
#include "caf/fwd.hpp"
#include "caf/string_view.hpp"
#include "caf/telemetry/label.hpp"
namespace caf::telemetry {
/// A single metric, identified by the values it sets for the label dimensions.
class CAF_CORE_EXPORT metric {
public:
// -- constructors, destructors, and assignment operators --------------------
metric(std::vector<label> labels) : labels_(std::move(labels)) {
// nop
}
virtual ~metric();
// -- properties -------------------------------------------------------------
const auto& labels() const noexcept {
return labels_;
}
protected:
// -- member variables -------------------------------------------------------
std::vector<label> labels_;
};
} // namespace caf::telemetry
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright 2011-2020 Dominik Charousset *
* *
* Distributed under the terms and conditions of the BSD 3-Clause License or *
* (at your option) under the terms and conditions of the Boost Software *
* License 1.0. See accompanying files LICENSE and LICENSE_ALTERNATIVE. *
* *
* If you did not receive a copy of the license files, see *
* http://opensource.org/licenses/BSD-3-Clause and *
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#pragma once
#include <memory>
#include <string>
#include <vector>
#include "caf/detail/core_export.hpp"
#include "caf/fwd.hpp"
namespace caf::telemetry {
/// Manages a collection (family) of metrics. All children share the same
/// prefix, name, and label dimensions.
class CAF_CORE_EXPORT metric_family {
public:
// -- constructors, destructors, and assignment operators --------------------
metric_family(metric_type type, std::string prefix, std::string name,
std::vector<std::string> label_names, std::string helptext,
std::string unit, bool is_sum)
: type_(type),
prefix_(std::move(prefix)),
name_(std::move(name)),
label_names_(std::move(label_names)),
helptext_(std::move(helptext)),
unit_(std::move(unit)),
is_sum_(is_sum) {
// nop
}
metric_family(const metric_family&) = delete;
metric_family& operator=(const metric_family&) = delete;
virtual ~metric_family();
// -- properties -------------------------------------------------------------
auto type() const noexcept {
return type_;
}
const auto& prefix() const noexcept {
return prefix_;
}
const auto& name() const noexcept {
return name_;
}
const auto& label_names() const noexcept {
return label_names_;
}
const auto& helptext() const noexcept {
return helptext_;
}
const auto& unit() const noexcept {
return unit_;
}
auto is_sum() const noexcept {
return is_sum_;
}
private:
metric_type type_;
std::string prefix_;
std::string name_;
std::vector<std::string> label_names_;
std::string helptext_;
std::string unit_;
bool is_sum_;
};
} // namespace caf::telemetry
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright 2011-2020 Dominik Charousset *
* *
* Distributed under the terms and conditions of the BSD 3-Clause License or *
* (at your option) under the terms and conditions of the Boost Software *
* License 1.0. See accompanying files LICENSE and LICENSE_ALTERNATIVE. *
* *
* If you did not receive a copy of the license files, see *
* http://opensource.org/licenses/BSD-3-Clause and *
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#pragma once
#include <algorithm>
#include <initializer_list>
#include <memory>
#include <mutex>
#include "caf/span.hpp"
#include "caf/string_view.hpp"
#include "caf/telemetry/label.hpp"
#include "caf/telemetry/label_view.hpp"
#include "caf/telemetry/metric.hpp"
#include "caf/telemetry/metric_family.hpp"
#include "caf/telemetry/metric_impl.hpp"
namespace caf::telemetry {
template <class Type>
class metric_family_impl : public metric_family {
public:
using super = metric_family;
using impl_type = metric_impl<Type>;
using extra_setting_type = typename impl_type::family_setting;
template <class... Ts>
metric_family_impl(const settings* config, std::string prefix,
std::string name, std::vector<std::string> label_names,
std::string helptext, std::string unit, bool is_sum,
Ts&&... xs)
: super(Type::runtime_type, std::move(prefix), std::move(name),
std::move(label_names), std::move(helptext), std::move(unit),
is_sum),
config_(config),
extra_setting_(std::forward<Ts>(xs)...) {
// nop
}
template <class... Ts>
metric_family_impl(std::string prefix, std::string name,
std::vector<std::string> label_names, std::string helptext,
std::string unit, bool is_sum, Ts&&... xs)
: super(Type::runtime_type, std::move(prefix), std::move(name),
std::move(label_names), std::move(helptext), std::move(unit),
is_sum),
config_(nullptr),
extra_setting_(std::forward<Ts>(xs)...) {
// nop
}
Type* get_or_add(span<const label_view> labels) {
auto has_label_values = [labels](const auto& metric_ptr) {
const auto& metric_labels = metric_ptr->labels();
return std::is_permutation(metric_labels.begin(), metric_labels.end(),
labels.begin(), labels.end());
};
std::unique_lock<std::mutex> guard{mx_};
auto m = std::find_if(metrics_.begin(), metrics_.end(), has_label_values);
if (m == metrics_.end()) {
std::vector<label> cpy{labels.begin(), labels.end()};
std::sort(cpy.begin(), cpy.end());
std::unique_ptr<impl_type> ptr;
if constexpr (std::is_same<extra_setting_type, unit_t>::value)
ptr.reset(new impl_type(std::move(cpy)));
else
ptr.reset(new impl_type(std::move(cpy), config_, extra_setting_));
m = metrics_.emplace(m, std::move(ptr));
}
return std::addressof(m->get()->impl());
}
Type* get_or_add(std::initializer_list<label_view> labels) {
return get_or_add(span<const label_view>{labels.begin(), labels.size()});
}
// -- properties --
const extra_setting_type& extra_setting() const noexcept {
return extra_setting_;
}
const settings* config() const noexcept {
return config_;
}
template <class Collector>
void collect(Collector& collector) const {
std::unique_lock<std::mutex> guard{mx_};
for (auto& ptr : metrics_)
collector(this, ptr.get(), std::addressof(ptr->impl()));
}
private:
const settings* config_;
extra_setting_type extra_setting_;
mutable std::mutex mx_;
std::vector<std::unique_ptr<impl_type>> metrics_;
};
} // namespace caf::telemetry
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright 2011-2020 Dominik Charousset *
* *
* Distributed under the terms and conditions of the BSD 3-Clause License or *
* (at your option) under the terms and conditions of the Boost Software *
* License 1.0. See accompanying files LICENSE and LICENSE_ALTERNATIVE. *
* *
* If you did not receive a copy of the license files, see *
* http://opensource.org/licenses/BSD-3-Clause and *
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#pragma once
#include <string>
#include <utility>
#include <vector>
#include "caf/fwd.hpp"
#include "caf/telemetry/label.hpp"
#include "caf/telemetry/metric.hpp"
namespace caf::telemetry {
template <class Type>
class metric_impl : public metric {
public:
using family_setting = typename Type::family_setting;
template <class... Ts>
explicit metric_impl(std::vector<label> labels, Ts&&... xs)
: metric(std::move(labels)), impl_(this->labels_, std::forward<Ts>(xs)...) {
// nop
}
Type& impl() noexcept {
return impl_;
}
const Type& impl() const noexcept {
return impl_;
}
Type* impl_ptr() noexcept {
return &impl_;
}
const Type* impl_ptr() const noexcept {
return &impl_;
}
private:
Type impl_;
};
} // namespace caf::telemetry
This diff is collapsed.
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright 2011-2020 Dominik Charousset *
* *
* Distributed under the terms and conditions of the BSD 3-Clause License or *
* (at your option) under the terms and conditions of the Boost Software *
* License 1.0. See accompanying files LICENSE and LICENSE_ALTERNATIVE. *
* *
* If you did not receive a copy of the license files, see *
* http://opensource.org/licenses/BSD-3-Clause and *
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#pragma once
namespace caf::telemetry {
enum class metric_type : uint8_t {
dbl_counter,
int_counter,
dbl_gauge,
int_gauge,
dbl_histogram,
int_histogram,
};
} // namespace caf::telemetry
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright 2011-2020 Dominik Charousset *
* *
* Distributed under the terms and conditions of the BSD 3-Clause License or *
* (at your option) under the terms and conditions of the Boost Software *
* License 1.0. See accompanying files LICENSE and LICENSE_ALTERNATIVE. *
* *
* If you did not receive a copy of the license files, see *
* http://opensource.org/licenses/BSD-3-Clause and *
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#pragma once
#include <chrono>
#include "caf/telemetry/histogram.hpp"
namespace caf::telemetry {
/// Convenience helper for measuring durations such as latency using a histogram
/// with second resolution. The measurement starts when creating the objects and
/// finishes when the timer goes out of scope.
class timer {
public:
using clock_type = std::chrono::steady_clock;
explicit timer(dbl_histogram* h) : h_(h) {
if (h)
start_ = clock_type::now();
}
~timer() {
if (h_)
observe(h_, start_);
}
auto histogram_ptr() const noexcept {
return h_;
}
auto started() const noexcept {
return start_;
}
static void observe(dbl_histogram* h, clock_type::time_point start) {
using dbl_sec = std::chrono::duration<double>;
auto end = clock_type::now();
h->observe(std::chrono::duration_cast<dbl_sec>(end - start).count());
}
private:
dbl_histogram* h_;
clock_type::time_point start_;
};
} // namespace caf::telemetry
......@@ -48,7 +48,7 @@ actor_registry::~actor_registry() {
// nop
}
actor_registry::actor_registry(actor_system& sys) : running_(0), system_(sys) {
actor_registry::actor_registry(actor_system& sys) : system_(sys) {
// nop
}
......@@ -96,19 +96,19 @@ void actor_registry::erase(actor_id key) {
void actor_registry::inc_running() {
# if CAF_LOG_LEVEL >= CAF_LOG_LEVEL_DEBUG
auto value = ++running_;
auto value = ++*system_.base_metrics().running_actors;
CAF_LOG_DEBUG(CAF_ARG(value));
# else
++running_;
system_.base_metrics().running_actors->inc();
# endif
}
size_t actor_registry::running() const {
return running_.load();
return static_cast<size_t>(system_.base_metrics().running_actors->value());
}
void actor_registry::dec_running() {
size_t new_val = --running_;
size_t new_val = --*system_.base_metrics().running_actors;
if (new_val <= 1) {
std::unique_lock<std::mutex> guard(running_mtx_);
running_cv_.notify_all();
......@@ -120,8 +120,8 @@ void actor_registry::await_running_count_equal(size_t expected) const {
CAF_ASSERT(expected == 0 || expected == 1);
CAF_LOG_TRACE(CAF_ARG(expected));
std::unique_lock<std::mutex> guard{running_mtx_};
while (running_ != expected) {
CAF_LOG_DEBUG(CAF_ARG(running_.load()));
while (running() != expected) {
CAF_LOG_DEBUG(CAF_ARG(running()));
running_cv_.wait(guard);
}
}
......
......@@ -46,11 +46,9 @@ struct kvstate {
using topic_set = std::unordered_set<std::string>;
std::unordered_map<key_type, std::pair<mapped_type, subscriber_set>> data;
std::unordered_map<strong_actor_ptr, topic_set> subscribers;
static const char* name;
static inline const char* name = "caf.system.config-server";
};
const char* kvstate::name = "config_server";
behavior config_serv_impl(stateful_actor<kvstate>* self) {
CAF_LOG_TRACE("");
std::string wildcard = "*";
......@@ -148,11 +146,9 @@ behavior config_serv_impl(stateful_actor<kvstate>* self) {
// on another node, users can spwan actors remotely.
struct spawn_serv_state {
static const char* name;
static inline const char* name = "caf.system.spawn-server";
};
const char* spawn_serv_state::name = "spawn_server";
behavior spawn_serv_impl(stateful_actor<spawn_serv_state>* self) {
CAF_LOG_TRACE("");
return {
......@@ -200,13 +196,13 @@ actor_system::module::~module() {
const char* actor_system::module::name() const noexcept {
switch (id()) {
case scheduler:
return "Scheduler";
return "scheduler";
case middleman:
return "Middleman";
return "middleman";
case openssl_manager:
return "OpenSSL Manager";
return "openssl-manager";
case network_manager:
return "Network Manager";
return "metwork-manager";
default:
return "???";
}
......@@ -216,9 +212,58 @@ actor_system::networking_module::~networking_module() {
// nop
}
namespace {
auto make_base_metrics(telemetry::metric_registry& reg) {
return actor_system::base_metrics_t{
// Initialize the base metrics.
reg.counter_singleton("caf.system", "rejected-messages",
"Number of rejected messages.", "1", true),
reg.counter_singleton("caf.system", "processed-messages",
"Number of processed messages.", "1", true),
reg.gauge_singleton("caf.system", "running-actors",
"Number of currently running actors."),
reg.gauge_singleton("caf.system", "queued-messages",
"Number of messages in all mailboxes.", "1", true),
};
}
auto make_actor_metric_families(telemetry::metric_registry& reg) {
// Handling a single message generally should take microseconds. Going up to
// several milliseconds usually indicates a problem (or blocking operations)
// but may still be expected for very compute-intense tasks. Single messages
// that approach seconds to process most likely indicate a severe issue.
// Hence, the default bucket settings focus on micro- and milliseconds.
std::array<double, 9> default_buckets{{
.00001, // 10us
.0001, // 100us
.0005, // 500us
.001, // 1ms
.01, // 10ms
.1, // 100ms
.5, // 500ms
1., // 1s
5., // 5s
}};
return actor_system::actor_metric_families_t{
reg.histogram_family<double>(
"caf.actor", "processing-time", {"name"}, default_buckets,
"Time an actor needs to process messages.", "seconds"),
reg.histogram_family<double>(
"caf.actor", "mailbox-time", {"name"}, default_buckets,
"Time a message waits in the mailbox before processing.", "seconds"),
reg.gauge_family("caf.actor", "mailbox-size", {"name"},
"Number of messages in the mailbox."),
};
}
} // namespace
actor_system::actor_system(actor_system_config& cfg)
: profiler_(cfg.profiler),
ids_(0),
metrics_(cfg),
base_metrics_(make_base_metrics(metrics_)),
logger_(new caf::logger(*this), false),
registry_(*this),
groups_(*this),
......@@ -231,6 +276,17 @@ actor_system::actor_system(actor_system_config& cfg)
CAF_SET_LOGGER_SYS(this);
for (auto& hook : cfg.thread_hooks_)
hook->init(*this);
// Cache some configuration parameters for faster lookups at runtime.
using string_list = std::vector<std::string>;
if (auto lst = get_if<string_list>(&cfg,
"caf.metrics-filters.actors.includes"))
metrics_actors_includes_ = std::move(*lst);
if (auto lst = get_if<string_list>(&cfg,
"caf.metrics-filters.actors.excludes"))
metrics_actors_excludes_ = std::move(*lst);
if (!metrics_actors_includes_.empty())
actor_metric_families_ = make_actor_metric_families(metrics_);
// Spin up modules.
for (auto& f : cfg.module_factories) {
auto mod_ptr = f(*this);
modules_[mod_ptr->id()].reset(mod_ptr);
......@@ -344,11 +400,6 @@ actor_system::~actor_system() {
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.
scheduler::abstract_coordinator& actor_system::scheduler() {
using ptr = scheduler::abstract_coordinator*;
......
......@@ -28,6 +28,7 @@
#include "caf/detail/sync_request_bouncer.hpp"
#include "caf/invoke_message_result.hpp"
#include "caf/logger.hpp"
#include "caf/telemetry/timer.hpp"
namespace caf {
......@@ -68,9 +69,17 @@ void blocking_actor::enqueue(mailbox_element_ptr ptr, execution_unit*) {
CAF_LOG_SEND_EVENT(ptr);
auto mid = ptr->mid;
auto src = ptr->sender;
auto collects_metrics = getf(abstract_actor::collects_metrics_flag);
if (collects_metrics) {
ptr->set_enqueue_time();
metrics_.mailbox_size->inc();
}
// returns false if mailbox has been closed
if (!mailbox().synchronized_push_back(mtx_, cv_, std::move(ptr))) {
CAF_LOG_REJECT_EVENT();
home_system().base_metrics().rejected_messages->inc();
if (collects_metrics)
metrics_.mailbox_size->dec();
if (mid.is_request()) {
detail::sync_request_bouncer srb{exit_reason()};
srb(src, mid);
......@@ -85,7 +94,7 @@ mailbox_element* blocking_actor::peek_at_next_mailbox_element() {
}
const char* blocking_actor::name() const {
return "blocking_actor";
return "user.blocking-actor";
}
void blocking_actor::launch(execution_unit*, bool, bool hide) {
......@@ -215,15 +224,33 @@ blocking_actor::mailbox_visitor::operator()(mailbox_element& x) {
return visit(f, sres);
};
// Post-process the returned value from the function body.
if (!self->getf(abstract_actor::collects_metrics_flag)) {
auto result = body();
if (result == intrusive::task_result::skip) {
CAF_AFTER_PROCESSING(self, invoke_message_result::skipped);
CAF_LOG_SKIP_EVENT();
} else {
CAF_AFTER_PROCESSING(self, invoke_message_result::consumed);
CAF_LOG_FINALIZE_EVENT();
}
return result;
} else {
auto t0 = std::chrono::steady_clock::now();
auto mbox_time = x.seconds_until(t0);
auto result = body();
if (result == intrusive::task_result::skip) {
CAF_AFTER_PROCESSING(self, invoke_message_result::skipped);
CAF_LOG_SKIP_EVENT();
auto& builtins = self->builtin_metrics();
telemetry::timer::observe(builtins.processing_time, t0);
builtins.mailbox_time->observe(mbox_time);
builtins.mailbox_size->dec();
} else {
CAF_AFTER_PROCESSING(self, invoke_message_result::consumed);
CAF_LOG_FINALIZE_EVENT();
}
return result;
}
}
void blocking_actor::receive_impl(receive_cond& rcc, message_id mid,
......@@ -324,8 +351,14 @@ bool blocking_actor::cleanup(error&& fail_state, execution_unit* host) {
mailbox_.close();
// TODO: messages that are stuck in the cache can get lost
detail::sync_request_bouncer bounce{fail_state};
while (mailbox_.queue().new_round(1000, bounce).consumed_items)
; // nop
auto dropped = mailbox_.queue().new_round(1000, bounce).consumed_items;
while (dropped > 0) {
if (getf(abstract_actor::collects_metrics_flag)) {
auto val = static_cast<int64_t>(dropped);
metrics_.mailbox_size->dec(val);
}
dropped = mailbox_.queue().new_round(1000, bounce).consumed_items;
}
}
// Dispatch to parent's `cleanup` function.
return super::cleanup(std::move(fail_state), host);
......
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright 2011-2020 Dominik Charousset *
* *
* Distributed under the terms and conditions of the BSD 3-Clause License or *
* (at your option) under the terms and conditions of the Boost Software *
* License 1.0. See accompanying files LICENSE and LICENSE_ALTERNATIVE. *
* *
* If you did not receive a copy of the license files, see *
* http://opensource.org/licenses/BSD-3-Clause and *
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
// Based on work of others. Retrieved on 2020-07-01 from
// https://github.com/Genivia/ugrep/blob/d2fb133/src/glob.cpp.
// Original header / license:
/******************************************************************************\
* Copyright (c) 2019, Robert van Engelen, Genivia Inc. All rights reserved. *
* *
* Redistribution and use in source and binary forms, with or without *
* modification, are permitted provided that the following conditions are met: *
* *
* (1) Redistributions of source code must retain the above copyright notice, *
* this list of conditions and the following disclaimer. *
* *
* (2) Redistributions in binary form must reproduce the above copyright *
* notice, this list of conditions and the following disclaimer in the *
* documentation and/or other materials provided with the distribution. *
* *
* (3) The name of the author may not be used to endorse or promote products *
* derived from this software without specific prior written permission. *
* *
* THIS SOFTWARE IS PROVIDED BY THE AUTHOR ``AS IS'' AND ANY EXPRESS OR IMPLIED *
* WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES OF *
* MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO *
* EVENT SHALL THE AUTHOR BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, *
* SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, *
* PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; *
* OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, *
* WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR *
* OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF *
* ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. *
\******************************************************************************/
#include "caf/detail/glob_match.hpp"
#include <cstdio>
#include <cstring>
#include "caf/config.hpp"
namespace caf::detail {
namespace {
static constexpr char path_separator =
#ifdef CAF_WINDOWS
'\\'
#else
'/'
#endif
;
int utf8(const char** s);
// match text against glob, return true or false
bool match(const char* text, const char* glob) {
// to iteratively backtrack on *
const char* text1_backup = nullptr;
const char* glob1_backup = nullptr;
// to iteratively backtrack on **
const char* text2_backup = nullptr;
const char* glob2_backup = nullptr;
// match until end of text
while (*text != '\0') {
switch (*glob) {
case '*':
if (*++glob == '*') {
// trailing ** match everything after /
if (*++glob == '\0')
return true;
// ** followed by a / match zero or more directories
if (*glob != '/')
return false;
// iteratively backtrack on **
text1_backup = NULL;
glob1_backup = NULL;
text2_backup = text;
glob2_backup = ++glob;
continue;
}
// iteratively backtrack on *
text1_backup = text;
glob1_backup = glob;
continue;
case '?':
// match anything except /
if (*text == path_separator)
break;
utf8(&text);
glob++;
continue;
case '[': {
int chr = utf8(&text), last = 0x10ffff;
// match anything except /
if (chr == path_separator)
break;
bool matched = false;
bool reverse = glob[1] == '^' || glob[1] == '!';
// inverted character class
if (reverse)
glob++;
glob++;
// match character class
while (*glob != '\0' && *glob != ']')
if (last < 0x10ffff && *glob == '-' && glob[1] != ']'
&& glob[1] != '\0'
? chr <= utf8(&++glob) && chr >= last
: chr == (last = utf8(&glob)))
matched = true;
if (matched == reverse)
break;
if (*glob)
glob++;
continue;
}
case '\\':
// literal match \-escaped character
glob++;
[[fallthrough]];
default:
#ifdef CAF_WINDOWS
if (*glob != *text && !(*glob == '/' && *text == '\\'))
#else
if (*glob != *text)
#endif
break;
text++;
glob++;
continue;
}
if (glob1_backup != NULL && *text1_backup != path_separator) {
// backtrack on the last *, do not jump over /
text = ++text1_backup;
glob = glob1_backup;
continue;
}
if (glob2_backup != NULL) {
// backtrack on the last **
text = ++text2_backup;
glob = glob2_backup;
continue;
}
return false;
}
while (*glob == '*')
glob++;
return *glob == '\0';
}
int utf8(const char** s) {
int c1, c2, c3, c = (unsigned char) **s;
if (c != '\0')
(*s)++;
if (c < 0x80)
return c;
c1 = (unsigned char) **s;
if (c < 0xC0 || (c == 0xC0 && c1 != 0x80) || c == 0xC1 || (c1 & 0xC0) != 0x80)
return 0xFFFD;
if (c1 != '\0')
(*s)++;
c1 &= 0x3F;
if (c < 0xE0)
return (((c & 0x1F) << 6) | c1);
c2 = (unsigned char) **s;
if ((c == 0xE0 && c1 < 0x20) || (c2 & 0xC0) != 0x80)
return 0xFFFD;
if (c2 != '\0')
(*s)++;
c2 &= 0x3F;
if (c < 0xF0)
return (((c & 0x0F) << 12) | (c1 << 6) | c2);
c3 = (unsigned char) **s;
if (c3 != '\0')
(*s)++;
if ((c == 0xF0 && c1 < 0x10) || (c == 0xF4 && c1 >= 0x10) || c >= 0xF5
|| (c3 & 0xC0) != 0x80)
return 0xFFFD;
return (((c & 0x07) << 18) | (c1 << 12) | (c2 << 6) | (c3 & 0x3F));
}
} // namespace
// pathname or basename matching, returns true or false
bool glob_match(const char* str, const char* glob) {
// an empty glob matches nothing and an empty string matches no glob
if (glob[0] == '\0' || str[0] == '\0')
return false;
// if str starts with ./ then remove these pairs
while (str[0] == '.' && str[1] == path_separator)
str += 2;
// if str starts with / then remove it
if (str[0] == path_separator)
++str;
// a leading / in the glob means globbing the str after removing the /
if (glob[0] == '/')
++glob;
return match(str, glob);
}
} // namespace caf::detail
......@@ -27,6 +27,7 @@
#include "caf/binary_deserializer.hpp"
#include "caf/binary_serializer.hpp"
#include "caf/default_attachable.hpp"
#include "caf/detail/glob_match.hpp"
#include "caf/exit_reason.hpp"
#include "caf/logger.hpp"
#include "caf/resumable.hpp"
......@@ -35,6 +36,36 @@
namespace caf {
namespace {
local_actor::metrics_t make_instance_metrics(local_actor* self) {
const auto& sys = self->home_system();
const auto& includes = sys.metrics_actors_includes();
const auto& excludes = sys.metrics_actors_excludes();
const auto* name = self->name();
auto matches = [name](const std::string& glob) {
return detail::glob_match(name, glob.c_str());
};
if (includes.empty()
|| std::none_of(includes.begin(), includes.end(), matches)
|| std::any_of(excludes.begin(), excludes.end(), matches))
return {
nullptr,
nullptr,
nullptr,
};
self->setf(abstract_actor::collects_metrics_flag);
const auto& families = sys.actor_metric_families();
string_view sv{name, strlen(name)};
return {
families.processing_time_family->get_or_add({{"name", sv}}),
families.mailbox_time_family->get_or_add({{"name", sv}}),
families.mailbox_size_family->get_or_add({{"name", sv}}),
};
}
} // namespace
local_actor::local_actor(actor_config& cfg)
: monitorable_actor(cfg),
context_(cfg.host),
......@@ -59,6 +90,10 @@ void local_actor::on_destroy() {
}
}
void local_actor::setup_metrics() {
metrics_ = make_instance_metrics(this);
}
void local_actor::request_response_timeout(timespan timeout, message_id mid) {
CAF_LOG_TRACE(CAF_ARG(timeout) << CAF_ARG(mid));
if (timeout == infinite)
......@@ -113,7 +148,7 @@ void local_actor::send_exit(const strong_actor_ptr& dest, error reason) {
}
const char* local_actor::name() const {
return "actor";
return "user.local-actor";
}
error local_actor::save_state(serializer&, const unsigned int) {
......
......@@ -34,7 +34,7 @@
namespace caf {
const char* monitorable_actor::name() const {
return "monitorable_actor";
return "user.monitorable-actor";
}
void monitorable_actor::attach(attachable_ptr ptr) {
......
......@@ -21,14 +21,12 @@
#include "caf/actor_ostream.hpp"
#include "caf/actor_system_config.hpp"
#include "caf/config.hpp"
#include "caf/inbound_path.hpp"
#include "caf/to_string.hpp"
#include "caf/scheduler/abstract_coordinator.hpp"
#include "caf/detail/default_invoke_result_visitor.hpp"
#include "caf/detail/private_thread.hpp"
#include "caf/detail/sync_request_bouncer.hpp"
#include "caf/inbound_path.hpp"
#include "caf/scheduler/abstract_coordinator.hpp"
#include "caf/to_string.hpp"
using namespace std::string_literals;
......@@ -166,6 +164,11 @@ void scheduled_actor::enqueue(mailbox_element_ptr ptr, execution_unit* eu) {
CAF_LOG_SEND_EVENT(ptr);
auto mid = ptr->mid;
auto sender = ptr->sender;
auto collects_metrics = getf(abstract_actor::collects_metrics_flag);
if (collects_metrics) {
ptr->set_enqueue_time();
metrics_.mailbox_size->inc();
}
switch (mailbox().push_back(std::move(ptr))) {
case intrusive::inbox_result::unblocked_reader: {
CAF_LOG_ACCEPT_EVENT(true);
......@@ -184,6 +187,9 @@ void scheduled_actor::enqueue(mailbox_element_ptr ptr, execution_unit* eu) {
}
case intrusive::inbox_result::queue_closed: {
CAF_LOG_REJECT_EVENT();
home_system().base_metrics().rejected_messages->inc();
if (collects_metrics)
metrics_.mailbox_size->dec();
if (mid.is_request()) {
detail::sync_request_bouncer f{exit_reason()};
f(sender, mid);
......@@ -203,7 +209,7 @@ mailbox_element* scheduled_actor::peek_at_next_mailbox_element() {
// -- overridden functions of local_actor --------------------------------------
const char* scheduled_actor::name() const {
return "scheduled_actor";
return "user.scheduled-actor";
}
void scheduled_actor::launch(execution_unit* eu, bool lazy, bool hide) {
......@@ -252,8 +258,14 @@ bool scheduled_actor::cleanup(error&& fail_state, execution_unit* host) {
get_normal_queue().flush_cache();
get_urgent_queue().flush_cache();
detail::sync_request_bouncer bounce{fail_state};
while (mailbox_.queue().new_round(1000, bounce).consumed_items)
; // nop
auto dropped = mailbox_.queue().new_round(1000, bounce).consumed_items;
while (dropped > 0) {
if (getf(abstract_actor::collects_metrics_flag)) {
auto val = static_cast<int64_t>(dropped);
metrics_.mailbox_size->dec(val);
}
dropped = mailbox_.queue().new_round(1000, bounce).consumed_items;
}
}
// Dispatch to parent's `cleanup` function.
return super::cleanup(std::move(fail_state), host);
......@@ -292,6 +304,7 @@ intrusive::task_result
scheduled_actor::mailbox_visitor::operator()(size_t, upstream_queue&,
mailbox_element& x) {
CAF_ASSERT(x.content().match_elements<upstream_msg>());
return run(x, [&] {
self->current_mailbox_element(&x);
CAF_LOG_RECEIVE_EVENT((&x));
CAF_BEFORE_PROCESSING(self, x);
......@@ -301,6 +314,7 @@ scheduled_actor::mailbox_visitor::operator()(size_t, upstream_queue&,
CAF_AFTER_PROCESSING(self, invoke_message_result::consumed);
return ++handled_msgs < max_throughput ? intrusive::task_result::resume
: intrusive::task_result::stop_all;
});
}
namespace {
......@@ -326,8 +340,7 @@ struct downstream_msg_visitor {
CAF_ASSERT(
inptr->slots == dm.slots
|| (dm.slots.sender == 0 && dm.slots.receiver == inptr->slots.receiver));
// TODO: replace with `if constexpr` when switching to C++17
if (std::is_same<T, downstream_msg::close>::value
if constexpr (std::is_same<T, downstream_msg::close>::value
|| std::is_same<T, downstream_msg::forced_close>::value) {
inptr.reset();
qs_ref.erase_later(dm.slots.receiver);
......@@ -354,6 +367,7 @@ intrusive::task_result scheduled_actor::mailbox_visitor::operator()(
size_t, downstream_queue& qs, stream_slot,
policy::downstream_messages::nested_queue_type& q, mailbox_element& x) {
CAF_LOG_TRACE(CAF_ARG(x) << CAF_ARG(handled_msgs));
return run(x, [&, this] {
self->current_mailbox_element(&x);
CAF_LOG_RECEIVE_EVENT((&x));
CAF_BEFORE_PROCESSING(self, x);
......@@ -364,22 +378,26 @@ intrusive::task_result scheduled_actor::mailbox_visitor::operator()(
CAF_AFTER_PROCESSING(self, invoke_message_result::consumed);
return ++handled_msgs < max_throughput ? res
: intrusive::task_result::stop_all;
});
}
intrusive::task_result
scheduled_actor::mailbox_visitor::operator()(mailbox_element& x) {
CAF_LOG_TRACE(CAF_ARG(x) << CAF_ARG(handled_msgs));
return run(x, [&, this] {
switch (self->reactivate(x)) {
case activation_result::terminated:
return intrusive::task_result::stop;
case activation_result::success:
return ++handled_msgs < max_throughput ? intrusive::task_result::resume
return ++handled_msgs < max_throughput
? intrusive::task_result::resume
: intrusive::task_result::stop_all;
case activation_result::skipped:
return intrusive::task_result::skip;
default:
return intrusive::task_result::resume;
}
});
}
resumable::resume_result scheduled_actor::resume(execution_unit* ctx,
......@@ -402,17 +420,22 @@ resumable::resume_result scheduled_actor::resume(execution_unit* ctx,
set_stream_timeout(tout);
}
};
mailbox_visitor f{this, handled_msgs, max_throughput};
mailbox_visitor f{this, handled_msgs, max_throughput,
getf(abstract_actor::collects_metrics_flag)};
mailbox_element_ptr ptr;
// Timeout for calling `advance_streams`.
while (handled_msgs < max_throughput) {
CAF_LOG_DEBUG("start new DRR round");
// TODO: maybe replace '3' with configurable / adaptive value?
// Dispatch on the different message categories in our mailbox.
if (!mailbox_.new_round(3, f).consumed_items) {
auto consumed = mailbox_.new_round(3, f).consumed_items;
if (consumed == 0) {
reset_timeouts_if_needed();
if (mailbox().try_block())
return resumable::awaiting_message;
} else {
auto signed_val = static_cast<int64_t>(consumed);
home_system().base_metrics().processed_messages->inc(signed_val);
}
// Check whether the visitor left the actor without behavior.
if (finalize()) {
......
......@@ -55,6 +55,10 @@ public:
mh_(what->content());
}
void setup_metrics() {
// nop
}
private:
message_handler mh_;
};
......
This diff is collapsed.
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright 2011-2020 Dominik Charousset *
* *
* Distributed under the terms and conditions of the BSD 3-Clause License or *
* (at your option) under the terms and conditions of the Boost Software *
* License 1.0. See accompanying files LICENSE and LICENSE_ALTERNATIVE. *
* *
* If you did not receive a copy of the license files, see *
* http://opensource.org/licenses/BSD-3-Clause and *
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#include "caf/telemetry/label.hpp"
#include "caf/telemetry/label_view.hpp"
namespace caf::telemetry {
label::label(string_view name, string_view value) : name_length_(name.size()) {
str_.reserve(name.size() + value.size() + 1);
str_.insert(str_.end(), name.begin(), name.end());
str_ += '=';
str_.insert(str_.end(), value.begin(), value.end());
}
label::label(const label_view& view) : label(view.name(), view.value()) {
// nop
}
void label::value(string_view new_value) {
str_.erase(name_length_ + 1);
str_.insert(str_.end(), new_value.begin(), new_value.end());
}
int label::compare(const label& x) const noexcept {
return str_.compare(x.str());
}
std::string to_string(const label& x) {
return x.str();
}
} // namespace caf::telemetry
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright 2011-2020 Dominik Charousset *
* *
* Distributed under the terms and conditions of the BSD 3-Clause License or *
* (at your option) under the terms and conditions of the Boost Software *
* License 1.0. See accompanying files LICENSE and LICENSE_ALTERNATIVE. *
* *
* If you did not receive a copy of the license files, see *
* http://opensource.org/licenses/BSD-3-Clause and *
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#include "caf/telemetry/label_view.hpp"
#include "caf/telemetry/label.hpp"
namespace caf::telemetry {
int label_view::compare(const label_view& x) const noexcept {
auto cmp1 = name().compare(x.name());
return cmp1 != 0 ? cmp1 : value().compare(x.value());
}
int label_view::compare(const label& x) const noexcept {
auto cmp1 = name().compare(x.name());
return cmp1 != 0 ? cmp1 : value().compare(x.value());
}
std::string to_string(const label_view& x) {
std::string result;
result.reserve(x.name().size() + 1 + x.value().size());
result.insert(result.end(), x.name().begin(), x.name().end());
result += '=';
result.insert(result.end(), x.value().begin(), x.value().end());
return result;
}
} // namespace caf::telemetry
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright 2011-2020 Dominik Charousset *
* *
* Distributed under the terms and conditions of the BSD 3-Clause License or *
* (at your option) under the terms and conditions of the Boost Software *
* License 1.0. See accompanying files LICENSE and LICENSE_ALTERNATIVE. *
* *
* If you did not receive a copy of the license files, see *
* http://opensource.org/licenses/BSD-3-Clause and *
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#include "caf/telemetry/metric.hpp"
#include "caf/config.hpp"
#include "caf/telemetry/metric_family.hpp"
namespace caf::telemetry {
metric::~metric() {
// nop
}
} // namespace caf::telemetry
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright 2011-2020 Dominik Charousset *
* *
* Distributed under the terms and conditions of the BSD 3-Clause License or *
* (at your option) under the terms and conditions of the Boost Software *
* License 1.0. See accompanying files LICENSE and LICENSE_ALTERNATIVE. *
* *
* If you did not receive a copy of the license files, see *
* http://opensource.org/licenses/BSD-3-Clause and *
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#include "caf/telemetry/metric_family.hpp"
namespace caf::telemetry {
metric_family::~metric_family() {
// nop
}
} // namespace caf::telemetry
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright 2011-2020 Dominik Charousset *
* *
* Distributed under the terms and conditions of the BSD 3-Clause License or *
* (at your option) under the terms and conditions of the Boost Software *
* License 1.0. See accompanying files LICENSE and LICENSE_ALTERNATIVE. *
* *
* If you did not receive a copy of the license files, see *
* http://opensource.org/licenses/BSD-3-Clause and *
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#include "caf/telemetry/metric_registry.hpp"
#include "caf/actor_system_config.hpp"
#include "caf/config.hpp"
#include "caf/telemetry/dbl_gauge.hpp"
#include "caf/telemetry/int_gauge.hpp"
#include "caf/telemetry/metric_family_impl.hpp"
#include "caf/telemetry/metric_impl.hpp"
#include "caf/telemetry/metric_type.hpp"
namespace {
template <class Range1, class Range2>
bool equals(Range1&& xs, Range2&& ys) {
return std::equal(xs.begin(), xs.end(), ys.begin(), ys.end());
}
} // namespace
namespace caf::telemetry {
metric_registry::metric_registry() : config_(nullptr) {
// nop
}
metric_registry::metric_registry(const actor_system_config& cfg) {
config_ = get_if<settings>(&cfg, "caf.metrics");
}
metric_registry::~metric_registry() {
// nop
}
metric_family* metric_registry::fetch(const string_view& prefix,
const string_view& name) {
auto eq = [&](const auto& ptr) {
return ptr->prefix() == prefix && ptr->name() == name;
};
if (auto i = std::find_if(families_.begin(), families_.end(), eq);
i != families_.end())
return i->get();
return nullptr;
}
std::vector<std::string>
metric_registry::to_sorted_vec(span<const string_view> xs) {
std::vector<std::string> result;
if (!xs.empty()) {
result.reserve(xs.size());
for (auto x : xs)
result.emplace_back(to_string(x));
std::sort(result.begin(), result.end());
}
return result;
}
std::vector<std::string>
metric_registry::to_sorted_vec(span<const label_view> xs) {
std::vector<std::string> result;
if (!xs.empty()) {
result.reserve(xs.size());
for (auto x : xs)
result.emplace_back(to_string(x.name()));
std::sort(result.begin(), result.end());
}
return result;
}
void metric_registry::assert_properties(const metric_family* ptr,
metric_type type,
span<const string_view> label_names,
string_view unit, bool is_sum) {
auto labels_match = [&] {
const auto& xs = ptr->label_names();
const auto& ys = label_names;
return std::is_sorted(ys.begin(), ys.end())
? std::equal(xs.begin(), xs.end(), ys.begin(), ys.end())
: std::is_permutation(xs.begin(), xs.end(), ys.begin(), ys.end());
};
if (ptr->type() != type)
CAF_RAISE_ERROR("full name with different metric type found");
if (!labels_match())
CAF_RAISE_ERROR("full name with different label dimensions found");
if (ptr->unit() != unit)
CAF_RAISE_ERROR("full name with different unit found");
if (ptr->is_sum() != is_sum)
CAF_RAISE_ERROR("full name with different is-sum flag found");
}
namespace {
struct label_name_eq {
bool operator()(string_view x, string_view y) const noexcept {
return x == y;
}
bool operator()(string_view x, const label_view& y) const noexcept {
return x == y.name();
}
bool operator()(const label_view& x, string_view y) const noexcept {
return x.name() == y;
}
bool operator()(label_view x, label_view y) const noexcept {
return x == y;
}
};
} // namespace
void metric_registry::assert_properties(const metric_family* ptr,
metric_type type,
span<const label_view> labels,
string_view unit, bool is_sum) {
auto labels_match = [&] {
const auto& xs = ptr->label_names();
const auto& ys = labels;
label_name_eq eq;
return std::is_sorted(ys.begin(), ys.end())
? std::equal(xs.begin(), xs.end(), ys.begin(), ys.end(), eq)
: std::is_permutation(xs.begin(), xs.end(), ys.begin(), ys.end(),
eq);
};
if (ptr->type() != type)
CAF_RAISE_ERROR("full name with different metric type found");
if (!labels_match())
CAF_RAISE_ERROR("full name with different label dimensions found");
if (ptr->unit() != unit)
CAF_RAISE_ERROR("full name with different unit found");
if (ptr->is_sum() != is_sum)
CAF_RAISE_ERROR("full name with different is-sum flag found");
}
} // namespace caf::telemetry
......@@ -112,7 +112,7 @@ struct fixture {
# define NAMED_ACTOR_STATE(type) \
struct type##_state { \
const char* name = #type; \
static inline const char* name = #type; \
}
NAMED_ACTOR_STATE(bar);
......
......@@ -77,6 +77,10 @@ struct fixture {
}
};
auto make_new_round_result(size_t consumed_items, bool stop_all) {
return new_round_result{consumed_items, stop_all};
}
} // namespace
CAF_TEST_FIXTURE_SCOPE(drr_cached_queue_tests, fixture)
......@@ -108,12 +112,12 @@ CAF_TEST(new_round) {
fill(queue, 1, 2, 3, 4, 5, 6, 7, 8, 9);
// Allow f to consume 2, 4, and 6.
auto round_result = queue.new_round(3, f);
CAF_CHECK_EQUAL(round_result, make_new_round_result(true));
CAF_CHECK_EQUAL(round_result, make_new_round_result(3, false));
CAF_CHECK_EQUAL(fseq, "246");
CAF_CHECK_EQUAL(queue.deficit(), 0);
// Allow g to consume 1, 3, 5, and 7.
round_result = queue.new_round(4, g);
CAF_CHECK_EQUAL(round_result, make_new_round_result(true));
CAF_CHECK_EQUAL(round_result, make_new_round_result(4, false));
CAF_CHECK_EQUAL(gseq, "1357");
CAF_CHECK_EQUAL(queue.deficit(), 0);
}
......@@ -128,21 +132,21 @@ CAF_TEST(skipping) {
return task_result::resume;
};
CAF_MESSAGE("make a round on an empty queue");
CAF_CHECK_EQUAL(queue.new_round(10, f), make_new_round_result(false));
CAF_CHECK_EQUAL(queue.new_round(10, f), make_new_round_result(0, false));
CAF_MESSAGE("make a round on a queue with only odd numbers (skip all)");
fill(queue, 1, 3, 5);
CAF_CHECK_EQUAL(queue.new_round(10, f), make_new_round_result(false));
CAF_CHECK_EQUAL(queue.new_round(10, f), make_new_round_result(0, false));
CAF_MESSAGE("make a round on a queue with an even number at the front");
fill(queue, 2);
CAF_CHECK_EQUAL(queue.new_round(10, f), make_new_round_result(true));
CAF_CHECK_EQUAL(queue.new_round(10, f), make_new_round_result(1, false));
CAF_CHECK_EQUAL(seq, "2");
CAF_MESSAGE("make a round on a queue with an even number in between");
fill(queue, 7, 9, 4, 11, 13);
CAF_CHECK_EQUAL(queue.new_round(10, f), make_new_round_result(true));
CAF_CHECK_EQUAL(queue.new_round(10, f), make_new_round_result(1, false));
CAF_CHECK_EQUAL(seq, "24");
CAF_MESSAGE("make a round on a queue with an even number at the back");
fill(queue, 15, 17, 6);
CAF_CHECK_EQUAL(queue.new_round(10, f), make_new_round_result(true));
CAF_CHECK_EQUAL(queue.new_round(10, f), make_new_round_result(1, false));
CAF_CHECK_EQUAL(seq, "246");
}
......@@ -196,7 +200,7 @@ CAF_TEST(alternating_consumer) {
// sequences and no odd value to read after 7 is available.
fill(queue, 1, 2, 3, 4, 5, 6, 7, 8, 9);
auto round_result = queue.new_round(1000, h);
CAF_CHECK_EQUAL(round_result, make_new_round_result(true));
CAF_CHECK_EQUAL(round_result, make_new_round_result(8, false));
CAF_CHECK_EQUAL(seq, "21436587");
CAF_CHECK_EQUAL(queue.deficit(), 0);
CAF_CHECK_EQUAL(deep_to_string(queue.cache()), "[9]");
......
......@@ -76,6 +76,10 @@ struct fixture {
}
};
auto make_new_round_result(size_t consumed_items, bool stop_all) {
return new_round_result{consumed_items, stop_all};
}
} // namespace
CAF_TEST_FIXTURE_SCOPE(drr_queue_tests, fixture)
......@@ -111,22 +115,22 @@ CAF_TEST(new_round) {
};
// Allow f to consume 1, 2, and 3 with a leftover deficit of 1.
auto round_result = queue.new_round(7, f);
CAF_CHECK_EQUAL(round_result, make_new_round_result(true));
CAF_CHECK_EQUAL(round_result, make_new_round_result(3, false));
CAF_CHECK_EQUAL(seq, "123");
CAF_CHECK_EQUAL(queue.deficit(), 1);
// Allow f to consume 4 and 5 with a leftover deficit of 0.
round_result = queue.new_round(8, f);
CAF_CHECK_EQUAL(round_result, make_new_round_result(true));
CAF_CHECK_EQUAL(round_result, make_new_round_result(2, false));
CAF_CHECK_EQUAL(seq, "12345");
CAF_CHECK_EQUAL(queue.deficit(), 0);
// Allow f to consume 6 with a leftover deficit of 0 (queue is empty).
round_result = queue.new_round(1000, f);
CAF_CHECK_EQUAL(round_result, make_new_round_result(true));
CAF_CHECK_EQUAL(round_result, make_new_round_result(1, false));
CAF_CHECK_EQUAL(seq, "123456");
CAF_CHECK_EQUAL(queue.deficit(), 0);
// new_round on an empty queue does nothing.
round_result = queue.new_round(1000, f);
CAF_CHECK_EQUAL(round_result, make_new_round_result(false));
CAF_CHECK_EQUAL(round_result, make_new_round_result(0, false));
CAF_CHECK_EQUAL(seq, "123456");
CAF_CHECK_EQUAL(queue.deficit(), 0);
}
......
......@@ -133,6 +133,10 @@ struct fixture {
}
};
auto make_new_round_result(size_t consumed_items, bool stop_all) {
return new_round_result{consumed_items, stop_all};
}
} // namespace
CAF_TEST_FIXTURE_SCOPE(wdrr_fixed_multiplexed_queue_tests, fixture)
......@@ -146,19 +150,19 @@ CAF_TEST(new_round) {
// Allow f to consume 2 items per nested queue.
fetch_helper f;
auto round_result = queue.new_round(2, f);
CAF_CHECK_EQUAL(round_result, make_new_round_result(true));
CAF_CHECK_EQUAL(round_result, make_new_round_result(6, false));
CAF_CHECK_EQUAL(f.result, "0:3,0:6,1:1,1:4,2:2,2:5");
CAF_REQUIRE_EQUAL(queue.empty(), false);
// Allow f to consume one more item from each queue.
f.result.clear();
round_result = queue.new_round(1, f);
CAF_CHECK_EQUAL(round_result, make_new_round_result(true));
CAF_CHECK_EQUAL(round_result, make_new_round_result(3, false));
CAF_CHECK_EQUAL(f.result, "0:9,1:7,2:8");
CAF_REQUIRE_EQUAL(queue.empty(), false);
// Allow f to consume the remainder, i.e., 12.
f.result.clear();
round_result = queue.new_round(1000, f);
CAF_CHECK_EQUAL(round_result, make_new_round_result(true));
CAF_CHECK_EQUAL(round_result, make_new_round_result(1, false));
CAF_CHECK_EQUAL(f.result, "0:12");
CAF_REQUIRE_EQUAL(queue.empty(), true);
}
......
......@@ -44,12 +44,10 @@ behavior pong(stateful_actor<pong_state>*) {
}
struct ping_state {
static const char* name;
static inline const char* name = "ping";
bool had_first_timeout = false; // unused in ping_singleN functions
};
const char* ping_state::name = "ping";
using ping_actor = stateful_actor<ping_state>;
using fptr = behavior (*)(ping_actor*, bool*, const actor&);
......
......@@ -123,21 +123,19 @@ CAF_TEST(stateful actors without explicit name use the name of the parent) {
struct state {
// empty
};
test_name<state>("scheduled_actor");
test_name<state>("user.scheduled-actor");
}
CAF_TEST(states with C string names override the default name) {
struct state {
const char* name = "testee";
};
test_name<state>("testee");
}
namespace {
CAF_TEST(states with STL string names override the default name) {
struct state {
std::string name = "testee2";
};
test_name<state>("testee2");
struct named_state {
static inline const char* name = "testee";
};
} // namespace
CAF_TEST(states with static C string names override the default name) {
test_name<named_state>("testee");
}
CAF_TEST(states can accept constructor arguments and provide a behavior) {
......@@ -167,10 +165,9 @@ CAF_TEST(states can accept constructor arguments and provide a behavior) {
}
CAF_TEST(states optionally take the self pointer as first argument) {
struct state_type {
struct state_type : named_state {
event_based_actor* self;
int x;
const char* name = "testee";
state_type(event_based_actor* self, int x) : self(self), x(x) {
// nop
}
......@@ -190,10 +187,9 @@ CAF_TEST(states optionally take the self pointer as first argument) {
}
CAF_TEST(typed actors can use typed_actor_pointer as self pointer) {
struct state_type {
struct state_type : named_state {
using self_pointer = typed_adder_actor::pointer_view;
self_pointer self;
const char* name = "testee";
int value;
state_type(self_pointer self, int x) : self(self), value(x) {
// nop
......
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright 2011-2020 Dominik Charousset *
* *
* Distributed under the terms and conditions of the BSD 3-Clause License or *
* (at your option) under the terms and conditions of the Boost Software *
* License 1.0. See accompanying files LICENSE and LICENSE_ALTERNATIVE. *
* *
* If you did not receive a copy of the license files, see *
* http://opensource.org/licenses/BSD-3-Clause and *
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#define CAF_SUITE telemetry.collector.prometheus
#include "caf/telemetry/collector/prometheus.hpp"
#include "caf/test/dsl.hpp"
#include "caf/telemetry/metric_registry.hpp"
#include "caf/telemetry/metric_type.hpp"
using namespace caf;
using namespace caf::literals;
using namespace caf::telemetry;
namespace {
struct fixture {
collector::prometheus exporter;
metric_registry registry;
};
} // namespace
CAF_TEST_FIXTURE_SCOPE(prometheus_tests, fixture)
CAF_TEST(the Prometheus collector generates text output) {
auto fb = registry.gauge_family("foo", "bar", {},
"Some value without labels.", "seconds");
auto sv = registry.gauge_family("some", "value", {"a", "b"},
"Some (total) value with two labels.", "1",
true);
auto ov = registry.gauge_family("other", "value", {"x"}, "", "seconds", true);
std::vector<int64_t> upper_bounds{1, 2, 4};
auto sr = registry.histogram_family("some", "request-duration", {"x"},
upper_bounds, "Some help.", "seconds");
fb->get_or_add({})->value(123);
sv->get_or_add({{"a", "1"}, {"b", "2"}})->value(12);
sv->get_or_add({{"b", "1"}, {"a", "2"}})->value(21);
ov->get_or_add({{"x", "true"}})->value(31337);
auto h = sr->get_or_add({{"x", "get"}});
h->observe(3);
h->observe(4);
h->observe(7);
CAF_CHECK_EQUAL(exporter.collect_from(registry, 42),
R"(# HELP foo_bar_seconds Some value without labels.
# TYPE foo_bar_seconds gauge
foo_bar_seconds 123 42000
# HELP some_value_total Some (total) value with two labels.
# TYPE some_value_total gauge
some_value_total{a="1",b="2"} 12 42000
some_value_total{a="2",b="1"} 21 42000
# TYPE other_value_seconds_total gauge
other_value_seconds_total{x="true"} 31337 42000
# HELP some_request_duration_seconds Some help.
# TYPE some_request_duration_seconds histogram
some_request_duration_seconds_bucket{x="get",le="1"} 0 42000
some_request_duration_seconds_bucket{x="get",le="2"} 0 42000
some_request_duration_seconds_bucket{x="get",le="4"} 2 42000
some_request_duration_seconds_bucket{x="get",le="+Inf"} 1 42000
some_request_duration_seconds_sum{x="get"} 14 42000
some_request_duration_seconds_count{x="get"} 1 42000
)"_sv);
CAF_MESSAGE("multiple runs generate the same output");
std::string res1;
{
auto buf = exporter.collect_from(registry);
res1.assign(buf.begin(), buf.end());
}
CAF_CHECK_EQUAL(res1, exporter.collect_from(registry));
}
CAF_TEST_FIXTURE_SCOPE_END()
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright 2011-2020 Dominik Charousset *
* *
* Distributed under the terms and conditions of the BSD 3-Clause License or *
* (at your option) under the terms and conditions of the Boost Software *
* License 1.0. See accompanying files LICENSE and LICENSE_ALTERNATIVE. *
* *
* If you did not receive a copy of the license files, see *
* http://opensource.org/licenses/BSD-3-Clause and *
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#define CAF_SUITE telemetry.counter
#include "caf/telemetry/counter.hpp"
#include "caf/test/dsl.hpp"
using namespace caf;
CAF_TEST(double counters can only increment) {
telemetry::dbl_counter c;
CAF_MESSAGE("counters start at 0");
CAF_CHECK_EQUAL(c.value(), 0.0);
CAF_MESSAGE("counters are incrementable");
c.inc();
c.inc(2.0);
CAF_CHECK_EQUAL(c.value(), 3.0);
CAF_MESSAGE("users can create counters with custom start values");
CAF_CHECK_EQUAL(telemetry::dbl_counter{42.0}.value(), 42.0);
}
CAF_TEST(integer counters can only increment) {
telemetry::int_counter c;
CAF_MESSAGE("counters start at 0");
CAF_CHECK_EQUAL(c.value(), 0);
CAF_MESSAGE("counters are incrementable");
c.inc();
c.inc(2);
CAF_CHECK_EQUAL(c.value(), 3);
CAF_MESSAGE("integer counters also support operator++");
CAF_CHECK_EQUAL(++c, 4);
CAF_MESSAGE("users can create counters with custom start values");
CAF_CHECK_EQUAL(telemetry::int_counter{42}.value(), 42);
}
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright 2011-2020 Dominik Charousset *
* *
* Distributed under the terms and conditions of the BSD 3-Clause License or *
* (at your option) under the terms and conditions of the Boost Software *
* License 1.0. See accompanying files LICENSE and LICENSE_ALTERNATIVE. *
* *
* If you did not receive a copy of the license files, see *
* http://opensource.org/licenses/BSD-3-Clause and *
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#define CAF_SUITE telemetry.gauge
#include "caf/telemetry/gauge.hpp"
#include "caf/test/dsl.hpp"
using namespace caf;
CAF_TEST(double gauges can increment and decrement) {
telemetry::dbl_gauge g;
CAF_MESSAGE("gauges start at 0");
CAF_CHECK_EQUAL(g.value(), 0.0);
CAF_MESSAGE("gauges are incrementable");
g.inc();
g.inc(2.0);
CAF_CHECK_EQUAL(g.value(), 3.0);
CAF_MESSAGE("gauges are decrementable");
g.dec();
g.dec(5.0);
CAF_CHECK_EQUAL(g.value(), -3.0);
CAF_MESSAGE("gauges allow setting values");
g.value(42.0);
CAF_CHECK_EQUAL(g.value(), 42.0);
CAF_MESSAGE("users can create gauges with custom start values");
CAF_CHECK_EQUAL(telemetry::dbl_gauge{42.0}.value(), 42.0);
}
CAF_TEST(integer gauges can increment and decrement) {
telemetry::int_gauge g;
CAF_MESSAGE("gauges start at 0");
CAF_CHECK_EQUAL(g.value(), 0);
CAF_MESSAGE("gauges are incrementable");
g.inc();
g.inc(2);
CAF_CHECK_EQUAL(g.value(), 3);
CAF_MESSAGE("gauges are decrementable");
g.dec();
g.dec(5);
CAF_CHECK_EQUAL(g.value(), -3);
CAF_MESSAGE("gauges allow setting values");
g.value(42);
CAF_MESSAGE("integer gauges also support operator++");
CAF_CHECK_EQUAL(++g, 43);
CAF_MESSAGE("users can create gauges with custom start values");
CAF_CHECK_EQUAL(telemetry::int_gauge{42}.value(), 42);
}
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright 2011-2020 Dominik Charousset *
* *
* Distributed under the terms and conditions of the BSD 3-Clause License or *
* (at your option) under the terms and conditions of the Boost Software *
* License 1.0. See accompanying files LICENSE and LICENSE_ALTERNATIVE. *
* *
* If you did not receive a copy of the license files, see *
* http://opensource.org/licenses/BSD-3-Clause and *
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#define CAF_SUITE telemetry.histogram
#include "caf/telemetry/histogram.hpp"
#include "caf/test/dsl.hpp"
#include <cmath>
#include <limits>
#include "caf/telemetry/gauge.hpp"
using namespace caf;
using namespace caf::telemetry;
CAF_TEST(double histograms use infinity for the last bucket) {
dbl_histogram h1{.1, .2, .4, .8};
CAF_CHECK_EQUAL(h1.buckets().size(), 5u);
CAF_CHECK_EQUAL(h1.buckets().front().upper_bound, .1);
CAF_CHECK(std::isinf(h1.buckets().back().upper_bound));
CAF_CHECK_EQUAL(h1.sum(), 0.0);
}
CAF_TEST(integer histograms use int_max for the last bucket) {
using limits = std::numeric_limits<int64_t>;
int_histogram h1{1, 2, 4, 8};
CAF_CHECK_EQUAL(h1.buckets().size(), 5u);
CAF_CHECK_EQUAL(h1.buckets().front().upper_bound, 1);
CAF_CHECK_EQUAL(h1.buckets().back().upper_bound, limits::max());
CAF_CHECK_EQUAL(h1.sum(), 0);
}
CAF_TEST(histograms aggregate to buckets and keep a sum) {
int_histogram h1{2, 4, 8};
for (int64_t value = 1; value < 11; ++value)
h1.observe(value);
auto buckets = h1.buckets();
CAF_REQUIRE_EQUAL(buckets.size(), 4u);
CAF_CHECK_EQUAL(buckets[0].count.value(), 2); // 1, 2
CAF_CHECK_EQUAL(buckets[1].count.value(), 2); // 3, 4
CAF_CHECK_EQUAL(buckets[2].count.value(), 4); // 5, 6, 7, 8
CAF_CHECK_EQUAL(buckets[3].count.value(), 2); // 9, 10
CAF_CHECK_EQUAL(h1.sum(), 55);
}
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright 2011-2020 Dominik Charousset *
* *
* Distributed under the terms and conditions of the BSD 3-Clause License or *
* (at your option) under the terms and conditions of the Boost Software *
* License 1.0. See accompanying files LICENSE and LICENSE_ALTERNATIVE. *
* *
* If you did not receive a copy of the license files, see *
* http://opensource.org/licenses/BSD-3-Clause and *
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#define CAF_SUITE telemetry.label
#include "caf/telemetry/label.hpp"
#include "caf/test/dsl.hpp"
#include "caf/telemetry/label_view.hpp"
using namespace caf;
using namespace caf::telemetry;
namespace {
template <class T>
size_t hash(const T& x) {
std::hash<T> f;
return f(x);
}
} // namespace
CAF_TEST(labels wrap name and value) {
CAF_CHECK_EQUAL(to_string(label{"foo", "bar"}), "foo=bar");
label foobar{"foo", "bar"};
CAF_CHECK_EQUAL(foobar.name(), "foo");
CAF_CHECK_EQUAL(foobar.value(), "bar");
CAF_CHECK_EQUAL(foobar.str(), "foo=bar");
CAF_CHECK_EQUAL(to_string(foobar), "foo=bar");
CAF_CHECK_EQUAL(foobar, label("foo", "bar"));
CAF_CHECK_EQUAL(hash(foobar), hash(label("foo", "bar")));
}
CAF_TEST(labels are convertible from views) {
label foobar{"foo", "bar"};
label_view foobar_view{"foo", "bar"};
CAF_CHECK_EQUAL(foobar, foobar_view);
CAF_CHECK_EQUAL(foobar, label{foobar_view});
CAF_CHECK_EQUAL(foobar.name(), foobar_view.name());
CAF_CHECK_EQUAL(foobar.value(), foobar_view.value());
CAF_CHECK_EQUAL(to_string(foobar), to_string(foobar_view));
CAF_CHECK_EQUAL(hash(foobar), hash(foobar_view));
CAF_CHECK_EQUAL(hash(foobar), hash(label{foobar_view}));
}
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright 2011-2020 Dominik Charousset *
* *
* Distributed under the terms and conditions of the BSD 3-Clause License or *
* (at your option) under the terms and conditions of the Boost Software *
* License 1.0. See accompanying files LICENSE and LICENSE_ALTERNATIVE. *
* *
* If you did not receive a copy of the license files, see *
* http://opensource.org/licenses/BSD-3-Clause and *
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#define CAF_SUITE telemetry.metric_registry
#include "caf/telemetry/metric_registry.hpp"
#include "caf/test/dsl.hpp"
#include "caf/string_view.hpp"
#include "caf/telemetry/counter.hpp"
#include "caf/telemetry/gauge.hpp"
#include "caf/telemetry/label_view.hpp"
#include "caf/telemetry/metric_type.hpp"
using namespace caf;
using namespace caf::telemetry;
namespace {
struct test_collector {
std::string result;
template <class T>
void operator()(const metric_family* family, const metric* instance,
const counter<T>* wrapped) {
concat(family, instance);
result += std::to_string(wrapped->value());
}
void operator()(const metric_family* family, const metric* instance,
const dbl_gauge* wrapped) {
concat(family, instance);
result += std::to_string(wrapped->value());
}
void operator()(const metric_family* family, const metric* instance,
const int_gauge* wrapped) {
concat(family, instance);
result += std::to_string(wrapped->value());
}
template <class T>
void operator()(const metric_family* family, const metric* instance,
const histogram<T>* wrapped) {
concat(family, instance);
result += std::to_string(wrapped->sum());
}
void concat(const metric_family* family, const metric* instance) {
result += '\n';
result += family->prefix();
result += '.';
result += family->name();
if (family->unit() != "1") {
result += '.';
result += family->unit();
}
if (family->is_sum())
result += ".total";
if (!instance->labels().empty()) {
result += '{';
auto i = instance->labels().begin();
concat(*i++);
while (i != instance->labels().end()) {
result += ',';
concat(*i++);
}
result += '}';
}
result += ' ';
}
void concat(const label& lbl) {
result.insert(result.end(), lbl.name().begin(), lbl.name().end());
result += "=\"";
result.insert(result.end(), lbl.value().begin(), lbl.value().end());
result += '"';
}
};
struct fixture {
metric_registry registry;
test_collector collector;
};
} // namespace
CAF_TEST_FIXTURE_SCOPE(metric_registry_tests, fixture)
CAF_TEST(registries lazily create metrics) {
std::vector<int64_t> upper_bounds{1, 2, 4, 8};
auto f = registry.gauge_family("caf", "running-actors", {"var1", "var2"},
"How many actors are currently running?");
auto g = registry.histogram_family("caf", "response-time", {"var1", "var2"},
upper_bounds, "How long take requests?");
std::vector<label_view> v1{{"var1", "foo"}, {"var2", "bar"}};
std::vector<label_view> v1_reversed{{"var2", "bar"}, {"var1", "foo"}};
std::vector<label_view> v2{{"var1", "bar"}, {"var2", "foo"}};
std::vector<label_view> v2_reversed{{"var2", "foo"}, {"var1", "bar"}};
f->get_or_add(v1)->value(42);
f->get_or_add(v2)->value(23);
CAF_CHECK_EQUAL(f->get_or_add(v1)->value(), 42);
CAF_CHECK_EQUAL(f->get_or_add(v1_reversed)->value(), 42);
CAF_CHECK_EQUAL(f->get_or_add(v2)->value(), 23);
CAF_CHECK_EQUAL(f->get_or_add(v2_reversed)->value(), 23);
g->get_or_add(v1)->observe(3);
g->get_or_add(v2)->observe(7);
CAF_CHECK_EQUAL(g->get_or_add(v1)->sum(), 3);
CAF_CHECK_EQUAL(g->get_or_add(v1_reversed)->sum(), 3);
CAF_CHECK_EQUAL(g->get_or_add(v2)->sum(), 7);
CAF_CHECK_EQUAL(g->get_or_add(v2_reversed)->sum(), 7);
}
CAF_TEST(registries allow users to collect all registered metrics) {
auto fb = registry.gauge_family("foo", "bar", {},
"Some value without labels.", "seconds");
auto sv = registry.gauge_family("some", "value", {"a", "b"},
"Some (total) value with two labels.", "1",
true);
auto ov = registry.gauge_family("other", "value", {"x"},
"Some (total) seconds with one label.",
"seconds", true);
auto ra = registry.gauge_family("caf", "running-actors", {"node"},
"How many actors are running?");
auto ms = registry.gauge_family("caf", "mailbox-size", {"name"},
"How full is the mailbox?");
CAF_MESSAGE("the registry always returns the same family object");
CAF_CHECK_EQUAL(fb, registry.gauge_family("foo", "bar", {}, "", "seconds"));
CAF_CHECK_EQUAL(sv, registry.gauge_family("some", "value", {"a", "b"}, "",
"1", true));
CAF_CHECK_EQUAL(sv, registry.gauge_family("some", "value", {"b", "a"}, "",
"1", true));
CAF_MESSAGE("families always return the same metric object for given labels");
CAF_CHECK_EQUAL(fb->get_or_add({}), fb->get_or_add({}));
CAF_CHECK_EQUAL(sv->get_or_add({{"a", "1"}, {"b", "2"}}),
sv->get_or_add({{"b", "2"}, {"a", "1"}}));
CAF_MESSAGE("collectors can observe all metrics in the registry");
fb->get_or_add({})->inc(123);
sv->get_or_add({{"a", "1"}, {"b", "2"}})->value(12);
sv->get_or_add({{"b", "1"}, {"a", "2"}})->value(21);
ov->get_or_add({{"x", "true"}})->value(31337);
ra->get_or_add({{"node", "localhost"}})->value(42);
ms->get_or_add({{"name", "printer"}})->value(3);
ms->get_or_add({{"name", "parser"}})->value(12);
registry.collect(collector);
CAF_CHECK_EQUAL(collector.result, R"(
foo.bar.seconds 123
some.value.total{a="1",b="2"} 12
some.value.total{a="2",b="1"} 21
other.value.seconds.total{x="true"} 31337
caf.running-actors{node="localhost"} 42
caf.mailbox-size{name="printer"} 3
caf.mailbox-size{name="parser"} 12)");
}
CAF_TEST(buckets for histograms are configurable via runtime settings) {
auto bounds = [](auto&& buckets) {
std::vector<int64_t> result;
for (auto&& bucket : buckets)
result.emplace_back(bucket.upper_bound);
result.pop_back();
return result;
};
settings cfg;
std::vector<int64_t> default_upper_bounds{1, 2, 4, 8};
std::vector<int64_t> upper_bounds{1, 2, 3, 5, 7};
std::vector<int64_t> alternative_upper_bounds{10, 20, 30};
put(cfg, "caf.response-time.buckets", upper_bounds);
put(cfg, "caf.response-time.var1=foo.buckets", alternative_upper_bounds);
registry.config(&cfg);
auto hf = registry.histogram_family("caf", "response-time", {"var1", "var2"},
default_upper_bounds,
"How long take requests?");
CAF_CHECK_EQUAL(hf->config(), get_if<settings>(&cfg, "caf.response-time"));
CAF_CHECK_EQUAL(hf->extra_setting(), upper_bounds);
auto h1 = hf->get_or_add({{"var1", "bar"}, {"var2", "baz"}});
CAF_CHECK_EQUAL(bounds(h1->buckets()), upper_bounds);
auto h2 = hf->get_or_add({{"var1", "foo"}, {"var2", "bar"}});
CAF_CHECK_NOT_EQUAL(h1, h2);
CAF_CHECK_EQUAL(bounds(h2->buckets()), alternative_upper_bounds);
}
CAF_TEST(counter_instance is a shortcut for using the family manually) {
auto fptr = registry.counter_family("http", "requests", {"method"},
"Number of HTTP requests.", "seconds",
true);
auto count = fptr->get_or_add({{"method", "put"}});
auto count2
= registry.counter_instance("http", "requests", {{"method", "put"}},
"Number of HTTP requests.", "seconds", true);
CAF_CHECK_EQUAL(count, count2);
}
CAF_TEST_FIXTURE_SCOPE_END()
#define CHECK_CONTAINS(str) \
CAF_CHECK_NOT_EQUAL(collector.result.find(str), npos)
CAF_TEST(enabling actor metrics per config creates metric instances) {
actor_system_config cfg;
test_coordinator_fixture<>::init_config(cfg);
put(cfg.content, "caf.metrics-filters.actors.includes",
std::vector<std::string>{"caf.system.*"});
actor_system sys{cfg};
test_collector collector;
sys.metrics().collect(collector);
auto npos = std::string::npos;
CHECK_CONTAINS(R"(caf.actor.mailbox-size{name="caf.system.spawn-server"})");
CHECK_CONTAINS(R"(caf.actor.mailbox-size{name="caf.system.config-server"})");
}
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright 2011-2020 Dominik Charousset *
* *
* Distributed under the terms and conditions of the BSD 3-Clause License or *
* (at your option) under the terms and conditions of the Boost Software *
* License 1.0. See accompanying files LICENSE and LICENSE_ALTERNATIVE. *
* *
* If you did not receive a copy of the license files, see *
* http://opensource.org/licenses/BSD-3-Clause and *
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#define CAF_SUITE telemetry.timer
#include "caf/telemetry/timer.hpp"
#include "caf/test/dsl.hpp"
using namespace caf;
using namespace caf::telemetry;
CAF_TEST(timers observe how much time passes in a scope) {
dbl_histogram h1{1, 2, 4, 8};
CAF_MESSAGE("timers call observe() with the measured time");
{
timer t{&h1};
CAF_CHECK_EQUAL(t.histogram_ptr(), &h1);
CAF_CHECK_GREATER(t.started().time_since_epoch().count(), 0);
}
CAF_CHECK_GREATER(h1.sum(), 0.0);
CAF_MESSAGE("timers constructed with a nullptr have no effect");
{
timer t{nullptr};
CAF_CHECK_EQUAL(t.histogram_ptr(), nullptr);
}
}
......@@ -99,13 +99,13 @@ public:
// nop
}
void
before_sending(const local_actor& self, mailbox_element& element) override {
void before_sending(const local_actor& self,
mailbox_element& element) override {
element.tracing_id.reset(new dummy_tracing_data(self.name()));
}
void
before_sending_scheduled(const local_actor& self, actor_clock::time_point,
void before_sending_scheduled(const local_actor& self,
actor_clock::time_point,
mailbox_element& element) override {
element.tracing_id.reset(new dummy_tracing_data(self.name()));
}
......@@ -154,7 +154,7 @@ const std::string& tracing_id(local_actor* self) {
# define NAMED_ACTOR_STATE(type) \
struct type##_state { \
const char* name = #type; \
static inline const char* name = #type; \
}
NAMED_ACTOR_STATE(alice);
......
......@@ -28,6 +28,7 @@ endfunction()
# -- add library target --------------------------------------------------------
add_library(libcaf_io_obj OBJECT ${CAF_IO_HEADERS}
src/detail/prometheus_broker.cpp
src/detail/socket_guard.cpp
src/io/abstract_broker.cpp
src/io/basp/header.cpp
......@@ -95,6 +96,7 @@ target_include_directories(caf-io-test PRIVATE "${CMAKE_CURRENT_SOURCE_DIR}/test
target_link_libraries(caf-io-test PRIVATE CAF::test)
caf_add_test_suites(caf-io-test
detail.prometheus_broker
io.basp.message_queue
io.basp_broker
io.broker
......
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright 2011-2020 Dominik Charousset *
* *
* Distributed under the terms and conditions of the BSD 3-Clause License or *
* (at your option) under the terms and conditions of the Boost Software *
* License 1.0. See accompanying files LICENSE and LICENSE_ALTERNATIVE. *
* *
* If you did not receive a copy of the license files, see *
* http://opensource.org/licenses/BSD-3-Clause and *
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#pragma once
#include <ctime>
#include <unordered_map>
#include <vector>
#include "caf/byte_buffer.hpp"
#include "caf/detail/io_export.hpp"
#include "caf/fwd.hpp"
#include "caf/io/broker.hpp"
#include "caf/telemetry/collector/prometheus.hpp"
namespace caf::detail {
/// Makes system metrics in the Prometheus format available via HTTP 1.1.
class CAF_IO_EXPORT prometheus_broker : public io::broker {
public:
explicit prometheus_broker(actor_config& cfg);
prometheus_broker(actor_config& cfg, io::doorman_ptr ptr);
~prometheus_broker() override;
const char* name() const override;
static bool has_process_metrics() noexcept;
behavior make_behavior() override;
private:
void scrape();
std::unordered_map<io::connection_handle, byte_buffer> requests_;
telemetry::collector::prometheus collector_;
time_t last_scrape_ = 0;
telemetry::dbl_gauge* cpu_time_ = nullptr;
telemetry::int_gauge* mem_size_ = nullptr;
telemetry::int_gauge* virt_mem_size_ = nullptr;
};
} // namespace caf::detail
......@@ -158,6 +158,9 @@ public:
/// Writes `data` into the buffer for a given connection.
void write(connection_handle hdl, size_t bs, const void* buf);
/// Writes `buf` into the buffer for a given connection.
void write(connection_handle hdl, span<const byte> buf);
/// Sends the content of the buffer for a given connection.
void flush(connection_handle hdl);
......
......@@ -24,7 +24,7 @@
namespace caf::io {
struct connection_helper_state {
static const char* name;
static inline const char* name = "caf.system.connection-helper";
};
CAF_IO_EXPORT behavior
......
......@@ -19,12 +19,15 @@
#pragma once
#include <chrono>
#include <list>
#include <map>
#include <memory>
#include <mutex>
#include <thread>
#include <vector>
#include "caf/actor_system.hpp"
#include "caf/config_value.hpp"
#include "caf/detail/io_export.hpp"
#include "caf/detail/unique_function.hpp"
#include "caf/expected.hpp"
......@@ -251,6 +254,13 @@ public:
std::forward<Ts>(xs)...);
}
/// Tries to open a port for exposing system metrics in the Prometheus text
/// format via HTTP.
/// @experimental
expected<uint16_t> expose_prometheus_metrics(uint16_t port,
const char* in = nullptr,
bool reuse = false);
/// Returns a middleman using the default network backend.
static actor_system::module* make(actor_system&, detail::type_list<>);
......@@ -313,6 +323,8 @@ private:
return system().spawn_class<Impl, Os>(cfg);
}
void expose_prometheus_metrics(const config_value::dictionary& cfg);
expected<strong_actor_ptr>
remote_spawn_impl(const node_id& nid, std::string& name, message& args,
std::set<std::string> s, timespan timeout);
......@@ -328,16 +340,27 @@ private:
static int exec_slave_mode(actor_system&, const actor_system_config&);
// environment
/// The actor environment.
actor_system& system_;
// prevents backend from shutting down unless explicitly requested
/// Prevents backend from shutting down unless explicitly requested.
network::multiplexer::supervisor_ptr backend_supervisor_;
// runs the backend
/// Runs the backend.
std::thread thread_;
// keeps track of "singleton-like" brokers
/// Keeps track of "singleton-like" brokers.
std::map<std::string, actor> named_brokers_;
// actor offering asynchronous IO by managing this singleton instance
/// Offers an asynchronous IO by managing this singleton instance.
middleman_actor manager_;
/// Protects `background_brokers_`.
mutable std::mutex background_brokers_mx_;
/// Stores hidden background actors that get killed automatically when the
/// actor systems shuts down.
std::list<actor> background_brokers_;
};
} // namespace caf::io
This diff is collapsed.
......@@ -22,6 +22,7 @@
#include "caf/make_counted.hpp"
#include "caf/none.hpp"
#include "caf/scheduler/abstract_coordinator.hpp"
#include "caf/span.hpp"
#include "caf/io/broker.hpp"
#include "caf/io/middleman.hpp"
......@@ -104,6 +105,10 @@ void abstract_broker::write(connection_handle hdl, size_t bs, const void* buf) {
out.insert(out.end(), first, last);
}
void abstract_broker::write(connection_handle hdl, span<const byte> buf) {
write(hdl, buf.size(), buf.data());
}
void abstract_broker::flush(connection_handle hdl) {
auto x = by_id(hdl);
if (x)
......@@ -361,7 +366,7 @@ abstract_broker::resume(execution_unit* ctx, size_t mt) {
}
const char* abstract_broker::name() const {
return "broker";
return "user.broker";
}
void abstract_broker::init_broker() {
......
......@@ -97,7 +97,7 @@ void basp_broker::on_exit() {
}
const char* basp_broker::name() const {
return "basp-broker";
return "caf.system.basp-broker";
}
behavior basp_broker::make_behavior() {
......
......@@ -38,8 +38,6 @@ auto autoconnect_timeout = std::chrono::minutes(10);
} // namespace
const char* connection_helper_state::name = "connection_helper";
behavior
connection_helper(stateful_actor<connection_helper_state>* self, actor b) {
CAF_LOG_TRACE(CAF_ARG(b));
......
......@@ -34,6 +34,7 @@
#include "caf/defaults.hpp"
#include "caf/detail/get_mac_addresses.hpp"
#include "caf/detail/get_root_uuid.hpp"
#include "caf/detail/prometheus_broker.hpp"
#include "caf/detail/ripemd_160.hpp"
#include "caf/detail/safe_equal.hpp"
#include "caf/detail/set_thread_name.hpp"
......@@ -302,10 +303,21 @@ void middleman::start() {
// Spawn utility actors.
auto basp = named_broker<basp_broker>("BASP");
manager_ = make_middleman_actor(system(), basp);
// Launch metrics exporters.
using dict = config_value::dictionary;
if (auto prom = get_if<dict>(&system().config(),
"caf.middleman.prometheus-http"))
expose_prometheus_metrics(*prom);
}
void middleman::stop() {
CAF_LOG_TRACE("");
{
std::unique_lock<std::mutex> guard{background_brokers_mx_};
for (auto& hdl : background_brokers_)
anon_send_exit(hdl, exit_reason::user_shutdown);
background_brokers_.clear();
}
backend().dispatch([=] {
CAF_LOG_TRACE("");
// managers_ will be modified while we are stopping each manager,
......@@ -345,7 +357,7 @@ void middleman::init(actor_system_config& cfg) {
cfg.set("middleman.attach-utility-actors", true)
.set("middleman.manual-multiplexing", true);
}
// add remote group module to config
// Add remote group module to config.
struct remote_groups : group_module {
public:
remote_groups(middleman& parent)
......@@ -379,10 +391,53 @@ void middleman::init(actor_system_config& cfg) {
// Compute and set ID for this network node.
auto this_node = node_id::default_data::local(cfg);
system().node_.swap(this_node);
// give config access to slave mode implementation
// Give config access to slave mode implementation.
cfg.slave_mode_fun = &middleman::exec_slave_mode;
}
expected<uint16_t> middleman::expose_prometheus_metrics(uint16_t port,
const char* in,
bool reuse) {
// Create the doorman for the broker.
doorman_ptr dptr;
if (auto maybe_dptr = backend().new_tcp_doorman(port, in, reuse))
dptr = std::move(*maybe_dptr);
else
return std::move(maybe_dptr.error());
auto actual_port = dptr->port();
// Spawn the actor and store its handle in background_brokers_.
using impl = detail::prometheus_broker;
actor_config cfg{&backend()};
auto hdl = system().spawn_impl<impl, hidden>(cfg, std::move(dptr));
std::list<actor> ls{std::move(hdl)};
std::unique_lock<std::mutex> guard{background_brokers_mx_};
background_brokers_.splice(background_brokers_.end(), ls);
return actual_port;
}
void middleman::expose_prometheus_metrics(const config_value::dictionary& cfg) {
// Read port, address and reuse flag from the config.
uint16_t port = 0;
if (auto cfg_port = get_if<uint16_t>(&cfg, "port")) {
port = *cfg_port;
} else {
CAF_LOG_ERROR("missing mandatory config field: "
"metrics.export.prometheus-http.port");
return;
}
const char* addr = nullptr;
if (auto cfg_addr = get_if<std::string>(&cfg, "address"))
if (*cfg_addr != "" && *cfg_addr != "0.0.0.0")
addr = cfg_addr->c_str();
auto reuse = get_or(cfg, "reuse", false);
if (auto res = expose_prometheus_metrics(port, addr, reuse)) {
CAF_LOG_INFO("expose Prometheus metrics at port" << *res);
} else {
CAF_LOG_ERROR("failed to expose Prometheus metrics:" << res.error());
return;
}
}
actor_system::module::id_t middleman::id() const {
return module::middleman;
}
......
......@@ -68,7 +68,7 @@ void middleman_actor_impl::on_exit() {
}
const char* middleman_actor_impl::name() const {
return "middleman_actor";
return "caf.system.middleman-actor";
}
auto middleman_actor_impl::make_behavior() -> behavior_type {
......
This diff is collapsed.
......@@ -947,9 +947,8 @@ T unbox(T* x) {
/// Implementation detail for `TESTEE` and `VARARGS_TESTEE`.
#define TESTEE_SCAFFOLD(tname) \
struct tname##_state : testee_state_base<tname##_state> { \
static const char* name; \
static inline const char* name = #tname; \
}; \
const char* tname##_state::name = #tname; \
using tname##_actor = stateful_actor<tname##_state>
/// Convenience macro for defining an actor named `tname`.
......
This diff is collapsed.
......@@ -26,6 +26,7 @@ Contents
ManagingGroupsOfWorkers
Streaming
Testing
Metrics
.. toctree::
:maxdepth: 2
......
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