Commit 39ad3c4f authored by Dominik Charousset's avatar Dominik Charousset

Implement better high-level API for Prometheus

parent b3520c82
...@@ -31,6 +31,7 @@ caf_add_component( ...@@ -31,6 +31,7 @@ caf_add_component(
SOURCES SOURCES
src/detail/convert_ip_endpoint.cpp src/detail/convert_ip_endpoint.cpp
src/detail/rfc6455.cpp src/detail/rfc6455.cpp
src/detail/shared_ssl_acceptor.cpp
src/net/abstract_actor_shell.cpp src/net/abstract_actor_shell.cpp
src/net/actor_shell.cpp src/net/actor_shell.cpp
src/net/binary/default_trait.cpp src/net/binary/default_trait.cpp
......
// This file is part of CAF, the C++ Actor Framework. See the file LICENSE in
// the main distribution directory for license terms and copyright or visit
// https://github.com/actor-framework/actor-framework/blob/master/LICENSE.
#pragma once
#include "caf/detail/net_export.hpp"
#include "caf/expected.hpp"
#include "caf/net/ssl/context.hpp"
#include "caf/net/ssl/fwd.hpp"
#include "caf/net/tcp_accept_socket.hpp"
#include <variant>
namespace caf::detail {
/// Like @ref net::ssl::acceptor but with a `shared_ptr` to the context.
class CAF_NET_EXPORT shared_ssl_acceptor {
public:
// -- member types -----------------------------------------------------------
using transport_type = net::ssl::transport;
// -- constructors, destructors, and assignment operators --------------------
shared_ssl_acceptor() = delete;
shared_ssl_acceptor(const shared_ssl_acceptor&) = default;
shared_ssl_acceptor& operator=(const shared_ssl_acceptor&) = default;
shared_ssl_acceptor(shared_ssl_acceptor&& other);
shared_ssl_acceptor& operator=(shared_ssl_acceptor&& other);
shared_ssl_acceptor(net::tcp_accept_socket fd,
std::shared_ptr<net::ssl::context> ctx)
: fd_(fd), ctx_(std::move(ctx)) {
// nop
}
// -- properties -------------------------------------------------------------
net::tcp_accept_socket fd() const noexcept {
return fd_;
}
net::ssl::context& ctx() noexcept {
return *ctx_;
}
const net::ssl::context& ctx() const noexcept {
return *ctx_;
}
private:
net::tcp_accept_socket fd_;
std::shared_ptr<net::ssl::context> ctx_;
};
// -- free functions -----------------------------------------------------------
/// Checks whether `acc` has a valid socket descriptor.
bool CAF_NET_EXPORT valid(const shared_ssl_acceptor& acc);
/// Closes the socket of `obj`.
void CAF_NET_EXPORT close(shared_ssl_acceptor& acc);
/// Tries to accept a new connection on `acc`. On success, wraps the new socket
/// into an SSL @ref connection and returns it.
expected<net::ssl::connection> CAF_NET_EXPORT accept(shared_ssl_acceptor& acc);
} // namespace caf::detail
// This file is part of CAF, the C++ Actor Framework. See the file LICENSE in
// the main distribution directory for license terms and copyright or visit
// https://github.com/actor-framework/actor-framework/blob/master/LICENSE.
#pragma once
#include "caf/actor_system.hpp"
#include "caf/defaults.hpp"
#include "caf/detail/accept_handler.hpp"
#include "caf/detail/connection_factory.hpp"
#include "caf/detail/shared_ssl_acceptor.hpp"
#include "caf/fwd.hpp"
#include "caf/net/http/server.hpp"
#include "caf/net/multiplexer.hpp"
#include "caf/net/prometheus/accept_factory.hpp"
#include "caf/net/prometheus/server.hpp"
#include "caf/net/ssl/transport.hpp"
#include "caf/net/stream_transport.hpp"
#include "caf/net/tcp_accept_socket.hpp"
#include "caf/none.hpp"
#include <cstdint>
#include <functional>
#include <variant>
namespace caf::detail {
template <class Transport>
class prometheus_conn_factory
: public connection_factory<typename Transport::connection_handle> {
public:
using state_ptr = net::prometheus::server::scrape_state_ptr;
using connection_handle = typename Transport::connection_handle;
explicit prometheus_conn_factory(state_ptr ptr) : ptr_(std::move(ptr)) {
// nop
}
net::socket_manager_ptr make(net::multiplexer* mpx,
connection_handle conn) override {
auto prom_serv = net::prometheus::server::make(ptr_);
auto http_serv = net::http::server::make(std::move(prom_serv));
auto transport = Transport::make(std::move(conn), std::move(http_serv));
return net::socket_manager::make(mpx, std::move(transport));
}
private:
state_ptr ptr_;
};
} // namespace caf::detail
namespace caf::net::prometheus {
class with_t;
/// Entry point for the `with(...).accept(...).start()` DSL.
class accept_factory {
public:
friend class with_t;
accept_factory(accept_factory&&) = default;
accept_factory(const accept_factory&) = delete;
accept_factory& operator=(accept_factory&&) noexcept = default;
accept_factory& operator=(const accept_factory&) noexcept = delete;
~accept_factory() {
if (auto* fd = std::get_if<tcp_accept_socket>(&state_))
close(*fd);
}
/// Configures how many concurrent connections we are allowing.
accept_factory& max_connections(size_t value) {
max_connections_ = value;
return *this;
}
/// Sets the callback for errors.
template <class F>
accept_factory& do_on_error(F callback) {
do_on_error_ = std::move(callback);
return *this;
}
/// Starts the Prometheus service in the background.
disposable start() {
switch (state_.index()) {
case 1: {
auto& cfg = std::get<1>(state_);
auto fd = make_tcp_accept_socket(cfg.port, cfg.address, cfg.reuse_addr);
if (fd)
return do_start(*fd);
if (do_on_error_)
do_on_error_(fd.error());
return {};
}
case 2: {
// Pass ownership of the socket to the accept handler.
auto fd = std::get<2>(state_);
state_ = none;
return do_start(fd);
}
default:
return {};
}
}
private:
struct config {
uint16_t port;
std::string address;
bool reuse_addr;
};
explicit accept_factory(actor_system* sys) : sys_(sys) {
// nop
}
disposable do_start(tcp_accept_socket fd) {
if (!ctx_) {
using factory_t = detail::prometheus_conn_factory<stream_transport>;
using impl_t = detail::accept_handler<tcp_accept_socket, stream_socket>;
auto mpx = &sys_->network_manager().mpx();
auto registry = &sys_->metrics();
auto state = prometheus::server::scrape_state::make(registry);
auto factory = std::make_unique<factory_t>(std::move(state));
auto impl = impl_t::make(fd, std::move(factory), max_connections_);
auto mgr = socket_manager::make(mpx, std::move(impl));
mpx->start(mgr);
return mgr->as_disposable();
}
using factory_t = detail::prometheus_conn_factory<ssl::transport>;
using acc_t = detail::shared_ssl_acceptor;
using impl_t = detail::accept_handler<acc_t, ssl::connection>;
auto mpx = &sys_->network_manager().mpx();
auto registry = &sys_->metrics();
auto state = prometheus::server::scrape_state::make(registry);
auto factory = std::make_unique<factory_t>(std::move(state));
auto impl = impl_t::make(acc_t{fd, ctx_}, std::move(factory),
max_connections_);
auto mgr = socket_manager::make(mpx, std::move(impl));
mpx->start(mgr);
return mgr->as_disposable();
}
void set_ssl(ssl::context ctx) {
ctx_ = std::make_shared<ssl::context>(std::move(ctx));
}
void init(uint16_t port, std::string address, bool reuse_addr) {
state_ = config{port, std::move(address), reuse_addr};
}
void init(tcp_accept_socket fd) {
state_ = fd;
}
/// Pointer to the hosting actor system.
actor_system* sys_;
/// Callback for errors.
std::function<void(const error&)> do_on_error_;
/// Configures the maximum number of concurrent connections.
size_t max_connections_ = defaults::net::max_connections.fallback;
/// User-defined state for getting things up and running.
std::variant<none_t, config, tcp_accept_socket> state_;
std::shared_ptr<ssl::context> ctx_;
};
} // namespace caf::net::prometheus
// This file is part of CAF, the C++ Actor Framework. See the file LICENSE in
// the main distribution directory for license terms and copyright or visit
// https://github.com/actor-framework/actor-framework/blob/master/LICENSE.
#pragma once
#include "caf/actor_system.hpp"
#include "caf/detail/accept_handler.hpp"
#include "caf/detail/connection_factory.hpp"
#include "caf/fwd.hpp"
#include "caf/net/http/server.hpp"
#include "caf/net/prometheus/server.hpp"
#include "caf/net/stream_transport.hpp"
namespace caf::detail {
template <class Transport>
class prometheus_conn_factory
: public connection_factory<typename Transport::connection_handle> {
public:
using state_ptr = net::prometheus::server::scrape_state_ptr;
using connection_handle = typename Transport::connection_handle;
explicit prometheus_conn_factory(state_ptr ptr) : ptr_(std::move(ptr)) {
// nop
}
net::socket_manager_ptr make(net::multiplexer* mpx,
connection_handle conn) override {
auto prom_serv = net::prometheus::server::make(ptr_);
auto http_serv = net::http::server::make(std::move(prom_serv));
auto transport = Transport::make(std::move(conn), std::move(http_serv));
return net::socket_manager::make(mpx, std::move(transport));
}
private:
state_ptr ptr_;
};
} // namespace caf::detail
namespace caf::net::prometheus {
/// Listens for incoming WebSocket connection on @p fd.
/// @param sys The host system.
/// @param fd An accept socket in listening mode. For a TCP socket, this socket
/// must already listen to a port.
template <class Transport = stream_transport, class Socket>
disposable serve(actor_system& sys, Socket fd, const settings& cfg = {}) {
using factory_t = detail::prometheus_conn_factory<Transport>;
using conn_t = typename Transport::connection_handle;
using impl_t = detail::accept_handler<Socket, conn_t>;
auto mpx = &sys.network_manager().mpx();
auto registry = &sys.metrics();
auto state = prometheus::server::scrape_state::make(registry);
auto factory = std::make_unique<factory_t>(std::move(state));
auto max_connections = get_or(cfg, defaults::net::max_connections);
auto impl = impl_t::make(fd, std::move(factory), max_connections);
auto mgr = socket_manager::make(mpx, std::move(impl));
mpx->start(mgr);
return mgr->as_disposable();
}
} // namespace caf::net::prometheus
// This file is part of CAF, the C++ Actor Framework. See the file LICENSE in
// the main distribution directory for license terms and copyright or visit
// https://github.com/actor-framework/actor-framework/blob/master/LICENSE.
#pragma once
#include "caf/fwd.hpp"
#include "caf/net/prometheus/accept_factory.hpp"
#include "caf/net/ssl/acceptor.hpp"
#include "caf/net/ssl/context.hpp"
#include "caf/net/tcp_accept_socket.hpp"
#include <cstdint>
namespace caf::net::prometheus {
/// Entry point for the `with(...).accept(...).start()` DSL.
class with_t {
public:
explicit with_t(actor_system* sys) : sys_(sys) {
// nop
}
with_t(const with_t&) noexcept = default;
with_t& operator=(const with_t&) noexcept = default;
/// Creates an `accept_factory` object for the given TCP `port` and
/// `bind_address`.
///
/// @param port Port number to bind to.
/// @param bind_address IP address to bind to. Default is an empty string.
/// @param reuse_addr Whether or not to set `SO_REUSEADDR`.
/// @returns an `accept_factory` object initialized with the given parameters.
accept_factory accept(uint16_t port, std::string bind_address = "",
bool reuse_addr = true) {
accept_factory factory{sys_};
factory.init(port, std::move(bind_address), std::move(reuse_addr));
return factory;
}
/// Creates an `accept_factory` object for the given accept socket.
///
/// @param fd File descriptor for the accept socket.
/// @returns an `accept_factory` object that will start a Prometheus server on
/// the given socket.
accept_factory accept(tcp_accept_socket fd) {
accept_factory factory{sys_};
factory.init(fd);
return factory;
}
/// Creates an `accept_factory` object for the given acceptor.
///
/// @param acc The SSL acceptor for incoming connections.
/// @returns an `accept_factory` object that will start a Prometheus server on
/// the given acceptor.
accept_factory accept(ssl::acceptor acc) {
accept_factory factory{sys_};
factory.set_ssl(std::move(std::move(acc.ctx())));
factory.init(acc.fd());
return factory;
}
/// Creates an `accept_factory` object for the given TCP `port` and
/// `bind_address`.
///
/// @param ctx The SSL context for encryption.
/// @param port Port number to bind to.
/// @param bind_address IP address to bind to. Default is an empty string.
/// @param reuse_addr Whether or not to set `SO_REUSEADDR`.
/// @returns an `accept_factory` object initialized with the given parameters.
accept_factory accept(ssl::context ctx, uint16_t port,
std::string bind_address = "", bool reuse_addr = true) {
accept_factory factory{sys_};
factory.set_ssl(std::move(std::move(ctx)));
factory.init(port, std::move(bind_address), std::move(reuse_addr));
return factory;
}
private:
/// Pointer to context.
actor_system* sys_;
};
inline with_t with(actor_system& sys) {
return with_t{&sys};
}
} // namespace caf::net::prometheus
...@@ -10,6 +10,8 @@ ...@@ -10,6 +10,8 @@
#include "caf/net/ssl/fwd.hpp" #include "caf/net/ssl/fwd.hpp"
#include "caf/net/tcp_accept_socket.hpp" #include "caf/net/tcp_accept_socket.hpp"
#include <variant>
namespace caf::net::ssl { namespace caf::net::ssl {
/// Wraps an accept socket and an SSL context. /// Wraps an accept socket and an SSL context.
......
// This file is part of CAF, the C++ Actor Framework. See the file LICENSE in
// the main distribution directory for license terms and copyright or visit
// https://github.com/actor-framework/actor-framework/blob/master/LICENSE.
#include "caf/detail/shared_ssl_acceptor.hpp"
#include "caf/expected.hpp"
#include "caf/net/ssl/connection.hpp"
#include "caf/net/tcp_stream_socket.hpp"
namespace caf::detail {
// -- constructors, destructors, and assignment operators ----------------------
shared_ssl_acceptor::shared_ssl_acceptor(shared_ssl_acceptor&& other)
: fd_(other.fd_), ctx_(std::move(other.ctx_)) {
other.fd_.id = net::invalid_socket_id;
}
shared_ssl_acceptor&
shared_ssl_acceptor::operator=(shared_ssl_acceptor&& other) {
fd_ = other.fd_;
ctx_ = std::move(other.ctx_);
other.fd_.id = net::invalid_socket_id;
return *this;
}
// -- free functions -----------------------------------------------------------
bool valid(const shared_ssl_acceptor& acc) {
return valid(acc.fd());
}
void close(shared_ssl_acceptor& acc) {
close(acc.fd());
}
expected<net::ssl::connection> accept(shared_ssl_acceptor& acc) {
auto fd = accept(acc.fd());
if (fd)
return acc.ctx().new_connection(*fd);
return expected<net::ssl::connection>{std::move(fd.error())};
}
} // namespace caf::detail
...@@ -7,7 +7,7 @@ ...@@ -7,7 +7,7 @@
#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/prometheus/with.hpp"
#include "caf/net/tcp_accept_socket.hpp" #include "caf/net/tcp_accept_socket.hpp"
#include "caf/net/tcp_stream_socket.hpp" #include "caf/net/tcp_stream_socket.hpp"
#include "caf/raise_error.hpp" #include "caf/raise_error.hpp"
...@@ -34,13 +34,12 @@ bool inspect(Inspector& f, prom_config& x) { ...@@ -34,13 +34,12 @@ bool inspect(Inspector& f, prom_config& x) {
} }
void launch_prom_server(actor_system& sys, const prom_config& cfg) { void launch_prom_server(actor_system& sys, const prom_config& cfg) {
if (auto fd = make_tcp_accept_socket(cfg.port, cfg.address, prometheus::with(sys)
cfg.reuse_address)) { .accept(cfg.port, cfg.address, cfg.reuse_address)
CAF_LOG_INFO("start Prometheus server at port" << local_port(*fd)); .do_on_error([](const error& err) {
prometheus::serve(sys, std::move(*fd)); CAF_LOG_WARNING("failed to start Prometheus server: " << err);
} else { })
CAF_LOG_WARNING("failed to start Prometheus server: " << fd.error()); .start();
}
} }
void launch_background_tasks(actor_system& sys) { void launch_background_tasks(actor_system& sys) {
......
...@@ -66,11 +66,10 @@ void close(acceptor& acc) { ...@@ -66,11 +66,10 @@ void close(acceptor& acc) {
} }
expected<connection> accept(acceptor& acc) { expected<connection> accept(acceptor& acc) {
if (auto fd = accept(acc.fd()); fd) { auto fd = accept(acc.fd());
if (fd)
return acc.ctx().new_connection(*fd); return acc.ctx().new_connection(*fd);
} else {
return expected<connection>{std::move(fd.error())}; return expected<connection>{std::move(fd.error())};
}
} }
} // namespace caf::net::ssl } // namespace caf::net::ssl
...@@ -561,6 +561,9 @@ caf.stream.output-buffer-size ...@@ -561,6 +561,9 @@ caf.stream.output-buffer-size
- **Type**: ``int_gauge`` - **Type**: ``int_gauge``
- **Label dimensions**: name, type. - **Label dimensions**: name, type.
.. _metrics_export:
Exporting Metrics to Prometheus Exporting Metrics to Prometheus
------------------------------- -------------------------------
......
...@@ -34,6 +34,13 @@ Contents ...@@ -34,6 +34,13 @@ Contents
Brokers Brokers
RemoteSpawn RemoteSpawn
.. toctree::
:maxdepth: 2
:caption: Networking Library
net/Overview
net/Prometheus
.. toctree:: .. toctree::
:maxdepth: 2 :maxdepth: 2
:caption: Appendix :caption: Appendix
......
.. _net_overview:
Overview
========
The networking module offers high-level APIs for individual protocols as well as
low-level building blocks for building implementing new protocols and assembling
protocol stacks.
High-level APIs
===============
The high-level APIs follow a factory pattern that configures each layer from the
bottom up, usually starting at the actor system. For example:
.. code-block:: C++
namespace ws = caf::net::web_socket;
auto conn = ws::with(sys)
.connect("localhost", 8080)
.start([&sys](auto pull, auto push) {
sys.spawn(my_actor, pull, push);
});
This code snippet tries to establish a WebSocket connection on
``localhost:8080`` and then spawns an actor that receives messages from the
WebSocket by reading from ``pull`` and sends messages to the WebSocket by
writing to ``push``. Errors are also transmitted over the ``pull`` resource.
A trivial implementation for ``my_actor`` that sends all messages it receives
back to the sender could look like this:
.. code-block:: C++
namespace ws = caf::net::web_socket;
void my_actor(caf::event_based_actor* self,
ws::default_trait::pull_resource pull,
ws::default_trait::push_resource push) {
self->make_observable().from_resource(pull).subscribe(push);
}
In general, connecting to a server follows the pattern:
.. code-block:: C++
PROTOCOL::with(CONTEXT).connect(WHERE).start(ON_CONNECT);
To establish the connection asynchronously:
.. code-block:: C++
PROTOCOL::with(CONTEXT).async_connect(WHERE).start(ON_CONNECT);
And for accepting incoming connections:
.. code-block:: C++
PROTOCOL::with(CONTEXT).accept(WHERE).start(ON_ACCEPT);
CAF also includes some self-contained services that users may simply start in
the background such as a Prometheus endpoint. These services take no callback in
``start``. For example:
.. code-block:: C++
prometheus::with(sys).accept(8008).start();
In all cases, CAF returns a ``disposable`` that allows users to cancel the
activity at any time. Note: when canceling a server, it only disposes the accept
socket itself. Previously established connections remain unaffected.
For error reporting, most factories allow setting a callback with
``do_on_error``.
Many protocols also accept additional configuration parameters. For example, the
factory for establishing WebSocket connections allows users to tweak parameters
for the handshake such as protocols or extensions fields. These options are
listed at the documentation for the individual protocols.
Layering
========
.. _net_prometheus:
Prometheus
==========
CAF ships a Prometheus server implementation that allows a scraper to collect
metrics from the actor system. The Prometheus server sits on top of an HTTP
layer.
Usually, users can simply use the configuration options of the actor system to
export metrics to scrapers: :ref:`metrics_export`. When setting these options,
CAF uses this implementation to start the Prometheus server in the background.
Starting a Prometheus Server
----------------------------
The Prometheus server always binds to the actor system, because it needs the
multiplexer and the metrics registry. Hence, the entry point is always:
.. code-block:: C++
caf::net::prometheus::with(sys)
From here, users can call ``accept`` with:
- A port, a bind address (defaults to the empty string for *any host*) and
whether to create the socket with ``SO_REUSEADDR`` (defaults to ``true``).
- An ``ssl::context``, a port, a bind address and whether to create the socket
with ``SO_REUSEADDR`` (with the same defaults as before).
- A ``tcp_accept_socket`` for accepting incoming connections.
- An ``ssl::acceptor`` for accepting incoming connections.
After setting up the factory, users may call these functions on the object:
- ``max_connections(size_t)`` for limiting the amount of concurrent connections
on the server.
- ``do_on_error(function<void(const caf::error&)>)`` for setting an error
callback.
- ``start()`` for initializing the server and running it in the background.
docutils
gitpython gitpython
sphinx_rtd_theme
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