Commit 02bb479c authored by Dominik Charousset's avatar Dominik Charousset

Spin up Prometheus from caf-net when available

parent ab21592f
...@@ -421,14 +421,17 @@ strong_actor_ptr middleman::remote_lookup(std::string name, ...@@ -421,14 +421,17 @@ strong_actor_ptr middleman::remote_lookup(std::string name,
void middleman::start() { void middleman::start() {
CAF_LOG_TRACE(""); CAF_LOG_TRACE("");
// Launch background tasks. // Launch background tasks unless caf-net is also available. In that case, the
if (auto prom = get_if<config_value::dictionary>( // net::middleman takes care of these.
&system().config(), "caf.middleman.prometheus-http")) { if (!system().has_network_manager()) {
auto ptr = std::make_unique<prometheus_scraping>(system()); if (auto prom = get_if<config_value::dictionary>(
if (auto port = ptr->start(*prom)) { &system().config(), "caf.middleman.prometheus-http")) {
CAF_ASSERT(*port != 0); auto ptr = std::make_unique<prometheus_scraping>(system());
prometheus_scraping_port_ = *port; if (auto port = ptr->start(*prom)) {
background_tasks_.emplace_back(std::move(ptr)); CAF_ASSERT(*port != 0);
prometheus_scraping_port_ = *port;
background_tasks_.emplace_back(std::move(ptr));
}
} }
} }
// Launch backend. // Launch backend.
......
...@@ -8,6 +8,7 @@ ...@@ -8,6 +8,7 @@
#include "caf/fwd.hpp" #include "caf/fwd.hpp"
#include "caf/net/http/server.hpp" #include "caf/net/http/server.hpp"
#include "caf/net/prometheus/server.hpp" #include "caf/net/prometheus/server.hpp"
#include "caf/net/stream_transport.hpp"
namespace caf::detail { namespace caf::detail {
......
...@@ -24,8 +24,8 @@ struct CAF_NET_EXPORT tcp_accept_socket : network_socket { ...@@ -24,8 +24,8 @@ struct CAF_NET_EXPORT tcp_accept_socket : network_socket {
/// Creates a new TCP socket to accept connections on a given port. /// Creates a new TCP socket to accept connections on a given port.
/// @param node The endpoint to listen on and the filter for incoming addresses. /// @param node The endpoint to listen on and the filter for incoming addresses.
/// Passing the address `0.0.0.0` will accept incoming connection from any host. /// Passing the address `0.0.0.0` will accept incoming connection
/// Passing port 0 lets the OS choose the port. /// from any host. Passing port 0 lets the OS choose the port.
/// @relates tcp_accept_socket /// @relates tcp_accept_socket
expected<tcp_accept_socket> expected<tcp_accept_socket>
CAF_NET_EXPORT make_tcp_accept_socket(ip_endpoint node, CAF_NET_EXPORT make_tcp_accept_socket(ip_endpoint node,
...@@ -33,14 +33,24 @@ expected<tcp_accept_socket> ...@@ -33,14 +33,24 @@ expected<tcp_accept_socket>
/// Creates a new TCP socket to accept connections on a given port. /// Creates a new TCP socket to accept connections on a given port.
/// @param node The endpoint to listen on and the filter for incoming addresses. /// @param node The endpoint to listen on and the filter for incoming addresses.
/// Passing the address `0.0.0.0` will accept incoming connection from any host. /// Passing the address `0.0.0.0` will accept incoming connection
/// Passing port 0 lets the OS choose the port. /// from any host. Passing port 0 lets the OS choose the port.
/// @param reuse_addr Optionally sets the SO_REUSEADDR option on the socket. /// @param reuse_addr Optionally sets the SO_REUSEADDR option on the socket.
/// @relates tcp_accept_socket /// @relates tcp_accept_socket
expected<tcp_accept_socket> expected<tcp_accept_socket>
CAF_NET_EXPORT make_tcp_accept_socket(const uri::authority_type& node, CAF_NET_EXPORT make_tcp_accept_socket(const uri::authority_type& node,
bool reuse_addr = false); bool reuse_addr = false);
/// Creates a new TCP socket to accept connections on a given port.
/// @param port The port for listening to incoming connection. Passing 0 lets
/// the OS choose a port.
/// @param addr The filter for incoming addresses. Passing the address `0.0.0.0`
/// will accept incoming connection from any host.
/// @param reuse_addr Optionally sets the SO_REUSEADDR option on the socket.
/// @relates tcp_accept_socket
expected<tcp_accept_socket> CAF_NET_EXPORT make_tcp_accept_socket(
uint16_t port, std::string addr = "0.0.0.0", bool reuse_addr = false);
/// Accepts a connection on `x`. /// Accepts a connection on `x`.
/// @param x Listening endpoint. /// @param x Listening endpoint.
/// @returns The socket that handles the accepted connection on success, an /// @returns The socket that handles the accepted connection on success, an
......
...@@ -7,6 +7,9 @@ ...@@ -7,6 +7,9 @@
#include "caf/actor_system_config.hpp" #include "caf/actor_system_config.hpp"
#include "caf/detail/set_thread_name.hpp" #include "caf/detail/set_thread_name.hpp"
#include "caf/expected.hpp" #include "caf/expected.hpp"
#include "caf/net/prometheus/serve.hpp"
#include "caf/net/tcp_accept_socket.hpp"
#include "caf/net/tcp_stream_socket.hpp"
#include "caf/raise_error.hpp" #include "caf/raise_error.hpp"
#include "caf/sec.hpp" #include "caf/sec.hpp"
#include "caf/send.hpp" #include "caf/send.hpp"
...@@ -14,6 +17,40 @@ ...@@ -14,6 +17,40 @@
namespace caf::net { namespace caf::net {
namespace {
struct prom_config {
uint16_t port;
std::string address = "0.0.0.0";
bool reuse_address = false;
};
template <class Inspector>
bool inspect(Inspector& f, prom_config& x) {
return f.object(x).fields(
f.field("port", x.port), f.field("address", x.address).fallback("0.0.0.0"),
f.field("reuse-address", x.reuse_address).fallback(false));
}
void launch_prom_server(actor_system& sys, const prom_config& cfg) {
if (auto fd = make_tcp_accept_socket(cfg.port, cfg.address,
cfg.reuse_address)) {
CAF_LOG_INFO("start Prometheus server at port" << local_port(*fd));
prometheus::serve(sys, std::move(*fd));
} else {
CAF_LOG_WARNING("failed to start Prometheus server: " << fd.error());
}
}
void launch_background_tasks(actor_system& sys) {
auto& cfg = sys.config();
if (auto pcfg = get_as<prom_config>(cfg, "caf.middleman.prometheus-http")) {
launch_prom_server(sys, *pcfg);
}
}
} // namespace
void middleman::init_global_meta_objects() { void middleman::init_global_meta_objects() {
// nop // nop
} }
...@@ -33,6 +70,7 @@ void middleman::start() { ...@@ -33,6 +70,7 @@ void middleman::start() {
detail::set_thread_name("caf.net.mpx"); detail::set_thread_name("caf.net.mpx");
sys_.thread_started(); sys_.thread_started();
mpx_.set_thread_id(); mpx_.set_thread_id();
launch_background_tasks(sys_);
mpx_.run(); mpx_.run();
sys_.thread_terminates(); sys_.thread_terminates();
}}; }};
...@@ -89,6 +127,9 @@ void middleman::add_module_options(actor_system_config& cfg) { ...@@ -89,6 +127,9 @@ void middleman::add_module_options(actor_system_config& cfg) {
"max. time between messages before declaring a node dead " "max. time between messages before declaring a node dead "
"(disabled if 0, ignored if heartbeats are disabled)") "(disabled if 0, ignored if heartbeats are disabled)")
.add<std::string>("network-backend", "legacy option"); .add<std::string>("network-backend", "legacy option");
config_option_adder{cfg.custom_options(), "caf.middleman.prometheus-http"}
.add<uint16_t>("port", "listening port for incoming scrapes")
.add<std::string>("address", "bind address for the HTTP server socket");
} }
} // namespace caf::net } // namespace caf::net
...@@ -121,6 +121,14 @@ make_tcp_accept_socket(const uri::authority_type& node, bool reuse_addr) { ...@@ -121,6 +121,14 @@ make_tcp_accept_socket(const uri::authority_type& node, bool reuse_addr) {
to_string(node)); to_string(node));
} }
expected<tcp_accept_socket>
make_tcp_accept_socket(uint16_t port, std::string addr, bool reuse_addr) {
uri::authority_type auth;
auth.port = port;
auth.host = std::move(addr);
return make_tcp_accept_socket(auth, reuse_addr);
}
expected<tcp_stream_socket> accept(tcp_accept_socket x) { expected<tcp_stream_socket> accept(tcp_accept_socket x) {
auto sock = ::accept(x.id, nullptr, nullptr); auto sock = ::accept(x.id, nullptr, nullptr);
if (sock == net::invalid_socket_id) { if (sock == net::invalid_socket_id) {
......
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