Commit fde0739a authored by Dominik Charousset's avatar Dominik Charousset

Add base metrics to the middleman

parent bfd26b48
...@@ -10,6 +10,10 @@ is based on [Keep a Changelog](https://keepachangelog.com). ...@@ -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 enum `caf::sec` received an additional error code: `connection_closed`.
- The new `byte_span` and `const_byte_span` aliases provide convenient - The new `byte_span` and `const_byte_span` aliases provide convenient
definitions when working with sequences of bytes. 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 ### Changed
......
...@@ -18,6 +18,7 @@ ...@@ -18,6 +18,7 @@
#pragma once #pragma once
#include <cstdint>
#include <vector> #include <vector>
#include "caf/actor_control_block.hpp" #include "caf/actor_control_block.hpp"
...@@ -29,10 +30,13 @@ ...@@ -29,10 +30,13 @@
#include "caf/detail/sync_request_bouncer.hpp" #include "caf/detail/sync_request_bouncer.hpp"
#include "caf/execution_unit.hpp" #include "caf/execution_unit.hpp"
#include "caf/io/basp/header.hpp" #include "caf/io/basp/header.hpp"
#include "caf/io/middleman.hpp"
#include "caf/logger.hpp" #include "caf/logger.hpp"
#include "caf/message.hpp" #include "caf/message.hpp"
#include "caf/message_id.hpp" #include "caf/message_id.hpp"
#include "caf/node_id.hpp" #include "caf/node_id.hpp"
#include "caf/telemetry/histogram.hpp"
#include "caf/telemetry/timer.hpp"
namespace caf::io::basp { namespace caf::io::basp {
...@@ -108,10 +112,15 @@ public: ...@@ -108,10 +112,15 @@ public:
CAF_LOG_ERROR("failed to read stages:" << source.get_error()); CAF_LOG_ERROR("failed to read stages:" << source.get_error());
return; return;
} }
auto& mm_metrics = ctx->system().middleman().metric_singletons;
auto t0 = telemetry::timer::clock_type::now();
if (!source.apply_objects(msg)) { if (!source.apply_objects(msg)) {
CAF_LOG_ERROR("failed to read message content:" << source.get_error()); CAF_LOG_ERROR("failed to read message content:" << source.get_error());
return; 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 // Intercept link messages. Forwarding actor proxies signalize linking
// by sending link_atom/unlink_atom message with src == dest. // by sending link_atom/unlink_atom message with src == dest.
if (auto view if (auto view
......
...@@ -48,6 +48,21 @@ class CAF_IO_EXPORT middleman : public actor_system::networking_module { ...@@ -48,6 +48,21 @@ class CAF_IO_EXPORT middleman : public actor_system::networking_module {
public: public:
friend class ::caf::actor_system; 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. /// Independent tasks that run in the background, usually in their own thread.
struct background_task { struct background_task {
virtual ~background_task(); virtual ~background_task();
...@@ -305,6 +320,9 @@ public: ...@@ -305,6 +320,9 @@ public:
return {}; return {};
} }
/// @private
metric_singletons_t metric_singletons;
protected: protected:
middleman(actor_system& sys); middleman(actor_system& sys);
......
...@@ -28,6 +28,8 @@ ...@@ -28,6 +28,8 @@
#include "caf/io/basp/version.hpp" #include "caf/io/basp/version.hpp"
#include "caf/io/basp/worker.hpp" #include "caf/io/basp/worker.hpp"
#include "caf/settings.hpp" #include "caf/settings.hpp"
#include "caf/telemetry/histogram.hpp"
#include "caf/telemetry/timer.hpp"
namespace caf::io::basp { namespace caf::io::basp {
...@@ -207,18 +209,24 @@ bool instance::dispatch(execution_unit* ctx, const strong_actor_ptr& sender, ...@@ -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, void instance::write(execution_unit* ctx, byte_buffer& buf, header& hdr,
payload_writer* pw) { payload_writer* pw) {
CAF_ASSERT(ctx != nullptr);
CAF_LOG_TRACE(CAF_ARG(hdr)); CAF_LOG_TRACE(CAF_ARG(hdr));
binary_serializer sink{ctx, buf}; binary_serializer sink{ctx, buf};
if (pw != nullptr) { if (pw != nullptr) {
// Write the BASP header after the payload. // Write the BASP header after the payload.
auto header_offset = buf.size(); auto header_offset = buf.size();
sink.skip(header_size); sink.skip(header_size);
auto& mm_metrics = ctx->system().middleman().metric_singletons;
auto t0 = telemetry::timer::clock_type::now();
if (!(*pw)(sink)) { if (!(*pw)(sink)) {
CAF_LOG_ERROR(sink.get_error()); CAF_LOG_ERROR(sink.get_error());
return; return;
} }
telemetry::timer::observe(mm_metrics.serialization_time, t0);
sink.seek(header_offset); sink.seek(header_offset);
auto payload_len = buf.size() - (header_offset + basp::header_size); 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); hdr.payload_len = static_cast<uint32_t>(payload_len);
} }
if (!sink.apply_objects(hdr)) if (!sink.apply_objects(hdr))
......
...@@ -67,6 +67,45 @@ namespace caf::io { ...@@ -67,6 +67,45 @@ namespace caf::io {
namespace { 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> template <class T>
class mm_impl : public middleman { class mm_impl : public middleman {
public: public:
...@@ -180,6 +219,7 @@ actor_system::module* middleman::make(actor_system& sys, detail::type_list<>) { ...@@ -180,6 +219,7 @@ actor_system::module* middleman::make(actor_system& sys, detail::type_list<>) {
middleman::middleman(actor_system& sys) : system_(sys) { middleman::middleman(actor_system& sys) : system_(sys) {
remote_groups_ = make_counted<detail::remote_group_module>(this); remote_groups_ = make_counted<detail::remote_group_module>(this);
metric_singletons = make_metrics(sys.metrics());
} }
expected<strong_actor_ptr> expected<strong_actor_ptr>
......
...@@ -27,6 +27,7 @@ ...@@ -27,6 +27,7 @@
#include "caf/actor_system.hpp" #include "caf/actor_system.hpp"
#include "caf/binary_serializer.hpp" #include "caf/binary_serializer.hpp"
#include "caf/io/basp/message_queue.hpp" #include "caf/io/basp/message_queue.hpp"
#include "caf/io/network/test_multiplexer.hpp"
#include "caf/make_actor.hpp" #include "caf/make_actor.hpp"
#include "caf/proxy_registry.hpp" #include "caf/proxy_registry.hpp"
...@@ -35,11 +36,20 @@ using namespace caf; ...@@ -35,11 +36,20 @@ using namespace caf;
namespace { namespace {
behavior testee_impl() { behavior testee_impl() {
return {[](ok_atom) { return {
[](ok_atom) {
// nop // nop
}}; },
};
} }
struct config : actor_system_config {
config() {
test_coordinator_fixture<>::init_config(*this);
load<io::middleman>();
}
};
class mock_actor_proxy : public actor_proxy { class mock_actor_proxy : public actor_proxy {
public: public:
explicit mock_actor_proxy(actor_config& cfg) : actor_proxy(cfg) { explicit mock_actor_proxy(actor_config& cfg) : actor_proxy(cfg) {
...@@ -74,7 +84,7 @@ private: ...@@ -74,7 +84,7 @@ private:
actor_system& sys_; actor_system& sys_;
}; };
struct fixture : test_coordinator_fixture<> { struct fixture : test_coordinator_fixture<config> {
detail::worker_hub<io::basp::worker> hub; detail::worker_hub<io::basp::worker> hub;
io::basp::message_queue queue; io::basp::message_queue queue;
mock_proxy_registry_backend proxies_backend; mock_proxy_registry_backend proxies_backend;
...@@ -87,6 +97,7 @@ struct fixture : test_coordinator_fixture<> { ...@@ -87,6 +97,7 @@ struct fixture : test_coordinator_fixture<> {
last_hop = unbox(std::move(tmp)); last_hop = unbox(std::move(tmp));
testee = sys.spawn<lazy_init>(testee_impl); testee = sys.spawn<lazy_init>(testee_impl);
sys.registry().put(testee.id(), testee); sys.registry().put(testee.id(), testee);
run();
} }
~fixture() { ~fixture() {
......
...@@ -433,7 +433,8 @@ configuration by the user. ...@@ -433,7 +433,8 @@ configuration by the user.
Base Metrics 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 caf.system.running-actors
- Tracks the current number of running actors in the system. - Tracks the current number of running actors in the system.
...@@ -451,6 +452,30 @@ caf.system.rejected-messages ...@@ -451,6 +452,30 @@ caf.system.rejected-messages
- **Type**: ``int_counter`` - **Type**: ``int_counter``
- **Label dimensions**: none. - **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 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