Commit 08a459df authored by Dominik Charousset's avatar Dominik Charousset

Allow exposing Prometheus metrics via config

parent fc66c8ac
...@@ -35,6 +35,8 @@ class CAF_IO_EXPORT prometheus_broker : public io::broker { ...@@ -35,6 +35,8 @@ class CAF_IO_EXPORT prometheus_broker : public io::broker {
public: public:
explicit prometheus_broker(actor_config& cfg); explicit prometheus_broker(actor_config& cfg);
prometheus_broker(actor_config& cfg, io::doorman_ptr ptr);
~prometheus_broker() override; ~prometheus_broker() override;
const char* name() const override; const char* name() const override;
......
...@@ -19,12 +19,15 @@ ...@@ -19,12 +19,15 @@
#pragma once #pragma once
#include <chrono> #include <chrono>
#include <list>
#include <map> #include <map>
#include <memory> #include <memory>
#include <mutex>
#include <thread> #include <thread>
#include <vector> #include <vector>
#include "caf/actor_system.hpp" #include "caf/actor_system.hpp"
#include "caf/config_value.hpp"
#include "caf/detail/io_export.hpp" #include "caf/detail/io_export.hpp"
#include "caf/detail/unique_function.hpp" #include "caf/detail/unique_function.hpp"
#include "caf/expected.hpp" #include "caf/expected.hpp"
...@@ -320,6 +323,8 @@ private: ...@@ -320,6 +323,8 @@ private:
return system().spawn_class<Impl, Os>(cfg); return system().spawn_class<Impl, Os>(cfg);
} }
void expose_prometheus_metrics(const config_value::dictionary& cfg);
expected<strong_actor_ptr> expected<strong_actor_ptr>
remote_spawn_impl(const node_id& nid, std::string& name, message& args, remote_spawn_impl(const node_id& nid, std::string& name, message& args,
std::set<std::string> s, timespan timeout); std::set<std::string> s, timespan timeout);
...@@ -335,16 +340,27 @@ private: ...@@ -335,16 +340,27 @@ private:
static int exec_slave_mode(actor_system&, const actor_system_config&); static int exec_slave_mode(actor_system&, const actor_system_config&);
// environment /// The actor environment.
actor_system& system_; 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_; network::multiplexer::supervisor_ptr backend_supervisor_;
// runs the backend
/// Runs the backend.
std::thread thread_; std::thread thread_;
// keeps track of "singleton-like" brokers
/// Keeps track of "singleton-like" brokers.
std::map<std::string, actor> named_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_; 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 } // namespace caf::io
...@@ -31,27 +31,14 @@ ...@@ -31,27 +31,14 @@
# define HAS_PROCESS_METRICS # define HAS_PROCESS_METRICS
namespace { namespace {
std::atomic<std::remove_pointer_t<vm_size_t>> global_page_size;
std::pair<int64_t, int64_t> get_mem_usage() { std::pair<int64_t, int64_t> get_mem_usage() {
mach_task_basic_info info; mach_task_basic_info info;
auto page_size = global_page_size.load();
if (page_size == 0) {
if (host_page_size(mach_host_self(), &page_size) == KERN_SUCCESS) {
global_page_size = page_size;
printf("page size = %d\n", (int) page_size);
} else {
puts("failed to get page size");
}
}
mach_msg_type_number_t count = MACH_TASK_BASIC_INFO_COUNT; mach_msg_type_number_t count = MACH_TASK_BASIC_INFO_COUNT;
if (task_info(mach_task_self(), MACH_TASK_BASIC_INFO, (task_info_t) &info, if (task_info(mach_task_self(), MACH_TASK_BASIC_INFO, (task_info_t) &info,
&count) &count)
!= KERN_SUCCESS) { != KERN_SUCCESS) {
puts("failed to get task info");
return {0, 0}; return {0, 0};
} }
puts("got task info");
return {static_cast<int64_t>(info.resident_size), return {static_cast<int64_t>(info.resident_size),
static_cast<int64_t>(info.virtual_size)}; static_cast<int64_t>(info.virtual_size)};
} }
...@@ -113,19 +100,24 @@ constexpr string_view request_ok = "HTTP/1.1 200 OK\r\n" ...@@ -113,19 +100,24 @@ constexpr string_view request_ok = "HTTP/1.1 200 OK\r\n"
} // namespace } // namespace
prometheus_broker::prometheus_broker(actor_config& cfg) : io::broker(cfg) { prometheus_broker::prometheus_broker(actor_config& cfg) : io::broker(cfg) {
#ifdef HAS_PROCESS_METRICS
using telemetry::dbl_gauge; using telemetry::dbl_gauge;
using telemetry::int_gauge; using telemetry::int_gauge;
auto& registry = system().telemetry(); auto& reg = system().telemetry();
#ifdef HAS_PROCESS_METRICS cpu_time_ = reg.singleton<dbl_gauge>(
cpu_time_ = registry.singleton<dbl_gauge>(
"process", "cpu", "Total user and system CPU time spent.", "seconds", true); "process", "cpu", "Total user and system CPU time spent.", "seconds", true);
mem_size_ = registry.singleton<int_gauge>( mem_size_ = reg.singleton<int_gauge>("process", "resident_memory",
"process", "resident_memory", " Resident memory size.", "bytes"); "Resident memory size.", "bytes");
virt_mem_size_ = registry.singleton<int_gauge>( virt_mem_size_ = reg.singleton<int_gauge>("process", "virtual_memory",
"process", "virtual_memory", " Virtual memory size.", "bytes"); "Virtual memory size.", "bytes");
#endif // HAS_PROCESS_METRICS #endif // HAS_PROCESS_METRICS
} }
prometheus_broker::prometheus_broker(actor_config& cfg, io::doorman_ptr ptr)
: prometheus_broker(cfg) {
add_doorman(std::move(ptr));
}
prometheus_broker::~prometheus_broker() { prometheus_broker::~prometheus_broker() {
// nop // nop
} }
...@@ -148,6 +140,7 @@ behavior prometheus_broker::make_behavior() { ...@@ -148,6 +140,7 @@ behavior prometheus_broker::make_behavior() {
auto& req = requests_[msg.handle]; auto& req = requests_[msg.handle];
if (req.size() + msg.buf.size() > max_request_size) { if (req.size() + msg.buf.size() > max_request_size) {
write(msg.handle, as_bytes(make_span(request_too_large))); write(msg.handle, as_bytes(make_span(request_too_large)));
flush(msg.handle);
close(msg.handle); close(msg.handle);
return; return;
} }
...@@ -161,6 +154,7 @@ behavior prometheus_broker::make_behavior() { ...@@ -161,6 +154,7 @@ behavior prometheus_broker::make_behavior() {
// Everything else, we ignore for now. // Everything else, we ignore for now.
if (!starts_with(req_str, "GET /metrics HTTP/1.")) { if (!starts_with(req_str, "GET /metrics HTTP/1.")) {
write(msg.handle, as_bytes(make_span(request_not_supported))); write(msg.handle, as_bytes(make_span(request_not_supported)));
flush(msg.handle);
close(msg.handle); close(msg.handle);
return; return;
} }
...@@ -172,6 +166,7 @@ behavior prometheus_broker::make_behavior() { ...@@ -172,6 +166,7 @@ behavior prometheus_broker::make_behavior() {
auto& dst = wr_buf(msg.handle); auto& dst = wr_buf(msg.handle);
dst.insert(dst.end(), hdr.begin(), hdr.end()); dst.insert(dst.end(), hdr.begin(), hdr.end());
dst.insert(dst.end(), payload.begin(), payload.end()); dst.insert(dst.end(), payload.begin(), payload.end());
flush(msg.handle);
close(msg.handle); close(msg.handle);
}, },
[=](const io::new_connection_msg& msg) { [=](const io::new_connection_msg& msg) {
......
...@@ -34,6 +34,7 @@ ...@@ -34,6 +34,7 @@
#include "caf/defaults.hpp" #include "caf/defaults.hpp"
#include "caf/detail/get_mac_addresses.hpp" #include "caf/detail/get_mac_addresses.hpp"
#include "caf/detail/get_root_uuid.hpp" #include "caf/detail/get_root_uuid.hpp"
#include "caf/detail/prometheus_broker.hpp"
#include "caf/detail/ripemd_160.hpp" #include "caf/detail/ripemd_160.hpp"
#include "caf/detail/safe_equal.hpp" #include "caf/detail/safe_equal.hpp"
#include "caf/detail/set_thread_name.hpp" #include "caf/detail/set_thread_name.hpp"
...@@ -302,10 +303,21 @@ void middleman::start() { ...@@ -302,10 +303,21 @@ void middleman::start() {
// Spawn utility actors. // Spawn utility actors.
auto basp = named_broker<basp_broker>("BASP"); auto basp = named_broker<basp_broker>("BASP");
manager_ = make_middleman_actor(system(), basp); manager_ = make_middleman_actor(system(), basp);
// Launch metrics exporters.
using dict = config_value::dictionary;
if (auto ex = get_if<dict>(&system().config(), "metrics.export"))
if (auto prom = get_if<dict>(ex, "prometheus-http"))
expose_prometheus_metrics(*prom);
} }
void middleman::stop() { void middleman::stop() {
CAF_LOG_TRACE(""); 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([=] { backend().dispatch([=] {
CAF_LOG_TRACE(""); CAF_LOG_TRACE("");
// managers_ will be modified while we are stopping each manager, // managers_ will be modified while we are stopping each manager,
...@@ -345,7 +357,7 @@ void middleman::init(actor_system_config& cfg) { ...@@ -345,7 +357,7 @@ void middleman::init(actor_system_config& cfg) {
cfg.set("middleman.attach-utility-actors", true) cfg.set("middleman.attach-utility-actors", true)
.set("middleman.manual-multiplexing", true); .set("middleman.manual-multiplexing", true);
} }
// add remote group module to config // Add remote group module to config.
struct remote_groups : group_module { struct remote_groups : group_module {
public: public:
remote_groups(middleman& parent) remote_groups(middleman& parent)
...@@ -379,10 +391,53 @@ void middleman::init(actor_system_config& cfg) { ...@@ -379,10 +391,53 @@ void middleman::init(actor_system_config& cfg) {
// Compute and set ID for this network node. // Compute and set ID for this network node.
auto this_node = node_id::default_data::local(cfg); auto this_node = node_id::default_data::local(cfg);
system().node_.swap(this_node); 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; 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 { actor_system::module::id_t middleman::id() const {
return module::middleman; return module::middleman;
} }
......
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