Commit ec3b75a1 authored by Dominik Charousset's avatar Dominik Charousset

Move Prometheus broker into its own thread

Metrics are vitally important for monitoring a distributed CAF
application. They should be available at all times, even if the CAF
middleman becomes overloaded. By moving the Prometheus broker to its
independent background thread, we shield it against overloads on the
regular middleman multiplexer.
parent e754ee26
......@@ -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;
......
......@@ -48,6 +48,14 @@ class CAF_IO_EXPORT middleman : public actor_system::networking_module {
public:
friend class ::caf::actor_system;
/// 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 +273,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);
......@@ -344,6 +345,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 +381,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_;
......
......@@ -264,12 +264,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();
},
};
}
......
......@@ -82,8 +82,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>();
}
......@@ -317,6 +377,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 +416,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 +433,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 +461,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 +487,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;
}
......
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