Commit 44c41ad2 authored by Dominik Charousset's avatar Dominik Charousset

Merge branch 'topic/neverlord/metrics'

parents e754ee26 fde0739a
......@@ -10,6 +10,10 @@ is based on [Keep a Changelog](https://keepachangelog.com).
- The enum `caf::sec` received an additional error code: `connection_closed`.
- The new `byte_span` and `const_byte_span` aliases provide convenient
definitions when working with sequences of bytes.
- The base metrics now include four new histograms for illuminating the I/O
module: `caf.middleman.inbound-messages-size`,
`caf.middleman.outbound-messages-size`, `caf.middleman.deserialization-time`
and `caf.middleman.serialization-time`.
### Changed
......
......@@ -299,15 +299,14 @@ void prometheus::append_histogram(const metric_family* family,
set_current_family(family, "histogram");
auto& vm = i->second;
auto buckets = val->buckets();
size_t index = 0;
for (; index < buckets.size() - 1; ++index) {
append(buf_, vm[index], buckets[index].count.value(), ' ',
ms_timestamp{now_}, '\n');
auto acc = ValueType{0};
auto index = size_t{0};
for (; index < buckets.size(); ++index) {
acc += buckets[index].count.value();
append(buf_, vm[index], acc, ' ', ms_timestamp{now_}, '\n');
}
auto count = buckets[index].count.value();
append(buf_, vm[index], count, ' ', ms_timestamp{now_}, '\n');
append(buf_, vm[++index], val->sum(), ' ', ms_timestamp{now_}, '\n');
append(buf_, vm[++index], count, ' ', ms_timestamp{now_}, '\n');
append(buf_, vm[index++], val->sum(), ' ', ms_timestamp{now_}, '\n');
append(buf_, vm[index++], acc, ' ', ms_timestamp{now_}, '\n');
}
} // namespace caf::telemetry::collector
......@@ -73,9 +73,9 @@ other_value_seconds_total{x="true"} 31337 42000
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_bucket{x="get",le="+Inf"} 3 42000
some_request_duration_seconds_sum{x="get"} 14 42000
some_request_duration_seconds_count{x="get"} 1 42000
some_request_duration_seconds_count{x="get"} 3 42000
)"_sv);
CAF_MESSAGE("multiple runs generate the same output");
std::string res1;
......
......@@ -352,10 +352,15 @@ public:
// -- observers --------------------------------------------------------------
/// Returns the number of open connections.
size_t num_connections() const {
size_t num_connections() const noexcept {
return scribes_.size();
}
/// Returns the number of attached doorman for accepting incoming connections.
size_t num_doormen() const noexcept {
return doormen_.size();
}
/// Returns all handles of all `scribe` instances attached to this broker.
std::vector<connection_handle> connections() const;
......
......@@ -18,6 +18,7 @@
#pragma once
#include <cstdint>
#include <vector>
#include "caf/actor_control_block.hpp"
......@@ -29,10 +30,13 @@
#include "caf/detail/sync_request_bouncer.hpp"
#include "caf/execution_unit.hpp"
#include "caf/io/basp/header.hpp"
#include "caf/io/middleman.hpp"
#include "caf/logger.hpp"
#include "caf/message.hpp"
#include "caf/message_id.hpp"
#include "caf/node_id.hpp"
#include "caf/telemetry/histogram.hpp"
#include "caf/telemetry/timer.hpp"
namespace caf::io::basp {
......@@ -108,10 +112,15 @@ public:
CAF_LOG_ERROR("failed to read stages:" << source.get_error());
return;
}
auto& mm_metrics = ctx->system().middleman().metric_singletons;
auto t0 = telemetry::timer::clock_type::now();
if (!source.apply_objects(msg)) {
CAF_LOG_ERROR("failed to read message content:" << source.get_error());
return;
}
telemetry::timer::observe(mm_metrics.deserialization_time, t0);
auto signed_size = static_cast<int64_t>(dref.payload_.size());
mm_metrics.inbound_messages_size->observe(signed_size);
// Intercept link messages. Forwarding actor proxies signalize linking
// by sending link_atom/unlink_atom message with src == dest.
if (auto view
......
......@@ -48,6 +48,29 @@ class CAF_IO_EXPORT middleman : public actor_system::networking_module {
public:
friend class ::caf::actor_system;
/// Metrics that the middleman collects by default.
struct metric_singletons_t {
/// Samples the size of inbound messages before deserializing them.
telemetry::int_histogram* inbound_messages_size = nullptr;
/// Samples how long the middleman needs to deserialize inbound messages.
telemetry::dbl_histogram* deserialization_time = nullptr;
/// Samples the size of outbound messages after serializing them.
telemetry::int_histogram* outbound_messages_size = nullptr;
/// Samples how long the middleman needs to serialize outbound messages.
telemetry::dbl_histogram* serialization_time = nullptr;
};
/// Independent tasks that run in the background, usually in their own thread.
struct background_task {
virtual ~background_task();
virtual bool start(const config_value::dictionary& cfg) = 0;
};
using background_task_ptr = std::unique_ptr<background_task>;
/// Adds message types of the I/O module to the global meta object table.
static void init_global_meta_objects();
......@@ -265,13 +288,6 @@ 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);
/// Adds module-specific options to the config before loading the module.
static void add_module_options(actor_system_config& cfg);
......@@ -304,6 +320,9 @@ public:
return {};
}
/// @private
metric_singletons_t metric_singletons;
protected:
middleman(actor_system& sys);
......@@ -344,6 +363,10 @@ private:
return system().spawn_class<Impl, Os>(cfg);
}
expected<uint16_t> expose_prometheus_metrics(uint16_t port,
const char* in = nullptr,
bool reuse = false);
void expose_prometheus_metrics(const config_value::dictionary& cfg);
expected<strong_actor_ptr>
......@@ -376,12 +399,8 @@ private:
/// 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_;
/// Handles to tasks that we spin up in start() and destroy in stop().
std::vector<background_task_ptr> background_tasks_;
/// Manages groups that run on a different node in the network.
detail::remote_group_module_ptr remote_groups_;
......
......@@ -225,11 +225,17 @@ bool prometheus_broker::has_process_metrics() noexcept {
behavior prometheus_broker::make_behavior() {
return {
[=](const io::new_data_msg& msg) {
auto flush_and_close = [this, &msg] {
flush(msg.handle);
close(msg.handle);
requests_.erase(msg.handle);
if (num_connections() + num_doormen() == 0)
quit();
};
auto& req = requests_[msg.handle];
if (req.size() + msg.buf.size() > max_request_size) {
write(msg.handle, as_bytes(make_span(request_too_large)));
flush(msg.handle);
close(msg.handle);
flush_and_close();
return;
}
req.insert(req.end(), msg.buf.begin(), msg.buf.end());
......@@ -242,8 +248,7 @@ behavior prometheus_broker::make_behavior() {
// Everything else, we ignore for now.
if (!starts_with(req_str, "GET /metrics HTTP/1.")) {
write(msg.handle, as_bytes(make_span(request_not_supported)));
flush(msg.handle);
close(msg.handle);
flush_and_close();
return;
}
// Collect metrics, ship response, and close.
......@@ -254,8 +259,7 @@ behavior prometheus_broker::make_behavior() {
auto& dst = wr_buf(msg.handle);
dst.insert(dst.end(), hdr.begin(), hdr.end());
dst.insert(dst.end(), payload.begin(), payload.end());
flush(msg.handle);
close(msg.handle);
flush_and_close();
},
[=](const io::new_connection_msg& msg) {
// Pre-allocate buffer for maximum request size.
......@@ -264,12 +268,14 @@ behavior prometheus_broker::make_behavior() {
configure_read(msg.handle, io::receive_policy::at_most(1024));
},
[=](const io::connection_closed_msg& msg) {
// No further action required other than cleaning up the state.
requests_.erase(msg.handle);
if (num_connections() + num_doormen() == 0)
quit();
},
[=](const io::acceptor_closed_msg&) {
// Shoud not happen.
quit(sec::socket_operation_failed);
CAF_LOG_ERROR("Prometheus Broker lost its acceptor!");
if (num_connections() + num_doormen() == 0)
quit();
},
};
}
......
......@@ -28,6 +28,8 @@
#include "caf/io/basp/version.hpp"
#include "caf/io/basp/worker.hpp"
#include "caf/settings.hpp"
#include "caf/telemetry/histogram.hpp"
#include "caf/telemetry/timer.hpp"
namespace caf::io::basp {
......@@ -207,18 +209,24 @@ bool instance::dispatch(execution_unit* ctx, const strong_actor_ptr& sender,
void instance::write(execution_unit* ctx, byte_buffer& buf, header& hdr,
payload_writer* pw) {
CAF_ASSERT(ctx != nullptr);
CAF_LOG_TRACE(CAF_ARG(hdr));
binary_serializer sink{ctx, buf};
if (pw != nullptr) {
// Write the BASP header after the payload.
auto header_offset = buf.size();
sink.skip(header_size);
auto& mm_metrics = ctx->system().middleman().metric_singletons;
auto t0 = telemetry::timer::clock_type::now();
if (!(*pw)(sink)) {
CAF_LOG_ERROR(sink.get_error());
return;
}
telemetry::timer::observe(mm_metrics.serialization_time, t0);
sink.seek(header_offset);
auto payload_len = buf.size() - (header_offset + basp::header_size);
auto signed_payload_len = static_cast<uint32_t>(payload_len);
mm_metrics.outbound_messages_size->observe(signed_payload_len);
hdr.payload_len = static_cast<uint32_t>(payload_len);
}
if (!sink.apply_objects(hdr))
......
......@@ -67,6 +67,45 @@ namespace caf::io {
namespace {
auto make_metrics(telemetry::metric_registry& reg) {
std::array<double, 9> default_time_buckets{{
.0002, // 20us
.0004, // 40us
.0006, // 60us
.0008, // 80us
.001, // 1ms
.005, // 5ms
.01, // 10ms
.05, // 50ms
.1, // 100ms
}};
std::array<int64_t, 9> default_size_buckets{{
100,
500,
1'000,
5'000,
10'000,
50'000,
100'000,
500'000,
1'000'000,
}};
return middleman::metric_singletons_t{
reg.histogram_singleton(
"caf.middleman", "inbound-messages-size", default_size_buckets,
"The size of inbound messages before deserializing them.", "bytes"),
reg.histogram_singleton<double>(
"caf.middleman", "deserialization-time", default_time_buckets,
"Time the middleman needs to deserialize inbound messages.", "seconds"),
reg.histogram_singleton(
"caf.middleman", "outbound-messages-size", default_size_buckets,
"The size of outbound messages after serializing them.", "bytes"),
reg.histogram_singleton<double>(
"caf.middleman", "serialization-time", default_time_buckets,
"Time the middleman needs to serialize outbound messages.", "seconds"),
};
}
template <class T>
class mm_impl : public middleman {
public:
......@@ -82,8 +121,68 @@ private:
T backend_;
};
class prometheus_scraping : public middleman::background_task {
public:
prometheus_scraping(actor_system& sys) : mpx_(&sys) {
// nop
}
bool start(const config_value::dictionary& cfg) override {
// 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 {
return false;
}
const char* addr = nullptr;
if (const std::string* 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 = start(port, addr, reuse)) {
CAF_LOG_INFO("expose Prometheus metrics at port" << *res);
return true;
} else {
CAF_LOG_ERROR("failed to expose Prometheus metrics:" << res.error());
return false;
}
}
expected<uint16_t> start(uint16_t port, const char* in, bool reuse) {
doorman_ptr dptr;
if (auto maybe_dptr = mpx_.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{&mpx_};
broker_ = mpx_.system().spawn_impl<impl, hidden>(cfg, std::move(dptr));
thread_ = std::thread{[this] { mpx_.run(); }};
return actual_port;
}
~prometheus_scraping() {
if (broker_) {
anon_send_exit(broker_, exit_reason::user_shutdown);
thread_.join();
}
}
private:
network::default_multiplexer mpx_;
actor broker_;
std::thread thread_;
};
} // namespace
middleman::background_task::~background_task() {
// nop
}
void middleman::init_global_meta_objects() {
caf::init_global_meta_objects<id_block::io_module>();
}
......@@ -120,6 +219,7 @@ actor_system::module* middleman::make(actor_system& sys, detail::type_list<>) {
middleman::middleman(actor_system& sys) : system_(sys) {
remote_groups_ = make_counted<detail::remote_group_module>(this);
metric_singletons = make_metrics(sys.metrics());
}
expected<strong_actor_ptr>
......@@ -317,6 +417,13 @@ strong_actor_ptr middleman::remote_lookup(std::string name,
void middleman::start() {
CAF_LOG_TRACE("");
// Launch background tasks.
if (auto prom = get_if<config_value::dictionary>(
&system().config(), "caf.middleman.prometheus-http")) {
auto ptr = std::make_unique<prometheus_scraping>(system());
if (ptr->start(*prom))
background_tasks_.emplace_back(std::move(ptr));
}
// Launch backend.
if (!get_or(config(), "caf.middleman.manual-multiplexing", false))
backend_supervisor_ = backend().make_supervisor();
......@@ -349,11 +456,6 @@ 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);
// Enable deserialization of groups.
system().groups().get_remote
= [this](const node_id& origin, const std::string& module_name,
......@@ -371,12 +473,6 @@ void middleman::start() {
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,
......@@ -405,6 +501,7 @@ void middleman::stop() {
if (!get_or(config(), "caf.middleman.attach-utility-actors", false))
self->wait_for(manager_);
destroy(manager_);
background_tasks_.clear();
}
void middleman::init(actor_system_config& cfg) {
......@@ -430,49 +527,6 @@ void middleman::init(actor_system_config& cfg) {
cfg.group_module_factories.emplace_back(dummy_fac);
}
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;
}
......
......@@ -27,6 +27,7 @@
#include "caf/actor_system.hpp"
#include "caf/binary_serializer.hpp"
#include "caf/io/basp/message_queue.hpp"
#include "caf/io/network/test_multiplexer.hpp"
#include "caf/make_actor.hpp"
#include "caf/proxy_registry.hpp"
......@@ -35,11 +36,20 @@ using namespace caf;
namespace {
behavior testee_impl() {
return {[](ok_atom) {
// nop
}};
return {
[](ok_atom) {
// nop
},
};
}
struct config : actor_system_config {
config() {
test_coordinator_fixture<>::init_config(*this);
load<io::middleman>();
}
};
class mock_actor_proxy : public actor_proxy {
public:
explicit mock_actor_proxy(actor_config& cfg) : actor_proxy(cfg) {
......@@ -74,7 +84,7 @@ private:
actor_system& sys_;
};
struct fixture : test_coordinator_fixture<> {
struct fixture : test_coordinator_fixture<config> {
detail::worker_hub<io::basp::worker> hub;
io::basp::message_queue queue;
mock_proxy_registry_backend proxies_backend;
......@@ -87,6 +97,7 @@ struct fixture : test_coordinator_fixture<> {
last_hop = unbox(std::move(tmp));
testee = sys.spawn<lazy_init>(testee_impl);
sys.registry().put(testee.id(), testee);
run();
}
~fixture() {
......
......@@ -433,7 +433,8 @@ configuration by the user.
Base Metrics
~~~~~~~~~~~~
The actor system collects this set of metrics always by default.
The actor system collects this set of metrics always by default (note that all
``caf.middleman`` metrics only appear when loading the I/O module).
caf.system.running-actors
- Tracks the current number of running actors in the system.
......@@ -451,6 +452,30 @@ caf.system.rejected-messages
- **Type**: ``int_counter``
- **Label dimensions**: none.
caf.middleman.inbound-messages-size
- Samples the size of inbound messages before deserializing them.
- **Type**: ``int_histogram``
- **Unit**: ``bytes``
- **Label dimensions**: none.
caf.middleman.outbound-messages-size
- Samples the size of outbound messages after serializing them.
- **Type**: ``int_histogram``
- **Unit**: ``bytes``
- **Label dimensions**: none.
caf.middleman.deserialization-time
- Samples how long the middleman needs to deserialize inbound messages.
- **Type**: ``dbl_histogram``
- **Unit**: ``seconds``
- **Label dimensions**: none.
caf.middleman.serialization-time
- Samples how long the middleman needs to serialize outbound messages.
- **Type**: ``dbl_histogram``
- **Unit**: ``seconds``
- **Label dimensions**: none.
Actor Metrics and Filters
~~~~~~~~~~~~~~~~~~~~~~~~~
......
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