Commit ab21592f authored by Dominik Charousset's avatar Dominik Charousset

Re-implement the Prometheus server in caf-net

With an actual HTTP backend in caf-net, we can implement the Prometheus
server in a much cleaner version than the previous ad-hoc version from
caf-io.
parent e94a5243
......@@ -51,6 +51,7 @@ caf_add_component(
src/net/network_socket.cpp
src/net/pipe_socket.cpp
src/net/pollset_updater.cpp
src/net/prometheus/server.cpp
src/net/socket.cpp
src/net/socket_event_layer.cpp
src/net/socket_manager.cpp
......@@ -94,6 +95,7 @@ caf_add_component(
net.operation
net.pipe_socket
net.producer_adapter
net.prometheus.server
net.socket
net.socket_guard
net.ssl.transport
......
// 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
namespace caf::net::http {
/// Stores context information for a request. For HTTP/2, this is the stream ID.
class context {};
} // namespace caf::net::http
......@@ -10,7 +10,6 @@
namespace caf::net::http {
class context;
class header;
class lower_layer;
class server;
......
......@@ -25,27 +25,26 @@ public:
virtual void suspend_reading() = 0;
/// Sends the next header to the client.
virtual bool
send_header(context ctx, status code, const header_fields_map& fields)
= 0;
virtual bool send_header(status code, const header_fields_map& fields) = 0;
/// Sends the payload after the header.
virtual bool send_payload(context, const_byte_span bytes) = 0;
virtual bool send_payload(const_byte_span bytes) = 0;
/// Sends a chunk of data if the full payload is unknown when starting to
/// send.
virtual bool send_chunk(context, const_byte_span bytes) = 0;
virtual bool send_chunk(const_byte_span bytes) = 0;
/// Sends the last chunk, completing a chunked payload.
virtual bool send_end_of_chunks() = 0;
/// Terminates an HTTP context.
virtual void fin(context) = 0;
/// Convenience function for sending header and payload. Automatically sets
/// the header fields `Content-Type` and `Content-Length`.
bool send_response(context ctx, status codek, std::string_view content_type,
bool send_response(status code, std::string_view content_type,
const_byte_span content);
/// @copydoc send_response
bool send_response(status code, std::string_view content_type,
std::string_view content);
};
} // namespace caf::net::http
......@@ -10,7 +10,6 @@
#include "caf/logger.hpp"
#include "caf/net/connection_acceptor.hpp"
#include "caf/net/fwd.hpp"
#include "caf/net/http/context.hpp"
#include "caf/net/http/header.hpp"
#include "caf/net/http/header_fields_map.hpp"
#include "caf/net/http/lower_layer.hpp"
......@@ -39,8 +38,6 @@ public:
using status_code_type = status;
using context_type = context;
using header_type = header;
enum class mode {
......@@ -88,17 +85,14 @@ public:
void suspend_reading() override;
bool send_header(context, status code,
const header_fields_map& fields) override;
bool send_header(status code, const header_fields_map& fields) override;
bool send_payload(context, const_byte_span bytes) override;
bool send_payload(const_byte_span bytes) override;
bool send_chunk(context, const_byte_span bytes) override;
bool send_chunk(const_byte_span bytes) override;
bool send_end_of_chunks() override;
void fin(context) override;
// -- stream_oriented::upper_layer implementation ----------------------------
error init(socket_manager* owner, stream_oriented::lower_layer* down,
......
......@@ -28,18 +28,12 @@ public:
= 0;
/// Consumes an HTTP message.
/// @param ctx Identifies this request. The response message must pass this
/// back to the lower layer. Allows clients to send multiple
/// requests in "parallel" (i.e., send multiple requests before
/// receiving the response on the first one).
/// @param hdr The header fields for the received message.
/// @param payload The payload of the received message.
/// @returns The number of consumed bytes or a negative value to signal an
/// error.
/// @note Discarded data is lost permanently.
virtual ptrdiff_t
consume(context ctx, const header& hdr, const_byte_span payload)
= 0;
virtual ptrdiff_t consume(const header& hdr, const_byte_span payload) = 0;
};
} // namespace caf::net::http
......@@ -89,9 +89,9 @@ public:
// -- thread-safe signaling --------------------------------------------------
/// Schedules a call to `mgr->handle_error(sec::discarded)`.
/// Schedules a call to `mgr->handle_error(sec::disposed)`.
/// @thread-safe
void discard(const socket_manager_ptr& mgr);
void dispose(const socket_manager_ptr& mgr);
/// Stops further reading by `mgr`.
/// @thread-safe
......@@ -224,7 +224,7 @@ private:
void do_register_reading(const socket_manager_ptr& mgr);
void do_discard(const socket_manager_ptr& mgr);
void do_dispose(const socket_manager_ptr& mgr);
void do_shutdown_reading(const socket_manager_ptr& mgr);
......
......@@ -26,7 +26,7 @@ public:
enum class code : uint8_t {
init_manager,
discard_manager,
dispose_manager,
shutdown_reading,
shutdown_writing,
run_action,
......
// 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/fwd.hpp"
#include "caf/net/http/server.hpp"
#include "caf/net/prometheus/server.hpp"
namespace caf::detail {
template <class Transport>
class prometheus_acceptor_factory {
public:
using state_ptr = net::prometheus::server::scrape_state_ptr;
explicit prometheus_acceptor_factory(state_ptr ptr) : ptr_(std::move(ptr)) {
// nop
}
error init(net::socket_manager*, const settings&) {
return none;
}
template <class Socket>
net::socket_manager_ptr make(net::multiplexer* mpx, Socket fd) {
auto prom_serv = net::prometheus::server::make(ptr_);
auto http_serv = net::http::server::make(std::move(prom_serv));
auto transport = Transport::make(fd, std::move(http_serv));
return net::socket_manager::make(mpx, fd, std::move(transport));
}
void abort(const error&) {
// nop
}
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) {
using factory_t = detail::prometheus_acceptor_factory<Transport>;
using impl_t = connection_acceptor<Socket, factory_t>;
auto mpx = &sys.network_manager().mpx();
auto registry = &sys.metrics();
auto state = prometheus::server::scrape_state::make(registry);
auto factory = factory_t{std::move(state)};
auto impl = impl_t::make(fd, 0, std::move(factory));
auto mgr = socket_manager::make(mpx, fd, std::move(impl));
mpx->init(mgr);
return mgr->make_disposer();
}
} // 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/http/fwd.hpp"
#include "caf/net/http/upper_layer.hpp"
#include "caf/telemetry/collector/prometheus.hpp"
#include "caf/telemetry/importer/process.hpp"
#include <chrono>
#include <memory>
#include <string_view>
namespace caf::net::prometheus {
/// Makes metrics available to clients via the Prometheus exposition format.
class server : public http::upper_layer {
public:
// -- member types -----------------------------------------------------------
/// State for scraping Metrics data. Shared between all server instances.
class scrape_state {
public:
using clock_type = std::chrono::steady_clock;
using time_point = clock_type::time_point;
using duration = clock_type::duration;
std::string_view scrape();
explicit scrape_state(telemetry::metric_registry* ptr)
: registry(ptr), last_scrape(duration{0}), proc_importer(*ptr) {
// nop
}
static std::shared_ptr<scrape_state> make(telemetry::metric_registry* ptr) {
return std::make_shared<scrape_state>(ptr);
}
telemetry::metric_registry* registry;
std::chrono::steady_clock::time_point last_scrape;
telemetry::importer::process proc_importer;
telemetry::collector::prometheus collector;
};
using scrape_state_ptr = std::shared_ptr<scrape_state>;
// -- factories --------------------------------------------------------------
static std::unique_ptr<server> make(scrape_state_ptr state) {
return std::unique_ptr<server>{new server(std::move(state))};
}
// -- implementation of http::upper_layer ------------------------------------
bool prepare_send() override;
bool done_sending() override;
void abort(const error& reason) override;
error init(socket_manager* owner, http::lower_layer* down,
const settings& config) override;
ptrdiff_t consume(const http::header& hdr, const_byte_span payload) override;
private:
explicit server(scrape_state_ptr state) : state_(std::move(state)) {
// nop
}
scrape_state_ptr state_;
http::lower_layer* down_ = nullptr;
};
} // namespace caf::net::prometheus
......@@ -74,6 +74,10 @@ public:
return make_actor_shell<Handle>(std::move(f));
}
/// Returns a thread-safe disposer for stopping the socket manager from an
/// outside context.
disposable make_disposer();
// -- properties -------------------------------------------------------------
/// Returns the handle for the managed socket.
......
......@@ -4,7 +4,6 @@
#include "caf/net/http/lower_layer.hpp"
#include "caf/net/http/context.hpp"
#include "caf/net/http/header_fields_map.hpp"
#include "caf/net/http/status.hpp"
......@@ -19,14 +18,18 @@ lower_layer::~lower_layer() {
// nop
}
bool lower_layer::send_response(context ctx, status code,
std::string_view content_type,
bool lower_layer::send_response(status code, std::string_view content_type,
const_byte_span content) {
auto content_size = std::to_string(content.size());
header_fields_map fields;
fields.emplace("Content-Type"sv, content_type);
fields.emplace("Content-Length"sv, content_size);
return send_header(ctx, code, fields) && send_payload(ctx, content);
return send_header(code, fields) && send_payload(content);
}
bool lower_layer::send_response(status code, std::string_view content_type,
std::string_view content) {
return send_response(code, content_type, as_bytes(make_span(content)));
}
} // namespace caf::net::http
......@@ -31,15 +31,14 @@ void server::suspend_reading() {
down_->configure_read(receive_policy::stop());
}
bool server::send_header(context, status code,
const header_fields_map& fields) {
bool server::send_header(status code, const header_fields_map& fields) {
down_->begin_output();
v1::write_header(code, fields, down_->output_buffer());
down_->end_output();
return true;
}
bool server::send_payload(context, const_byte_span bytes) {
bool server::send_payload(const_byte_span bytes) {
down_->begin_output();
auto& buf = down_->output_buffer();
buf.insert(buf.end(), bytes.begin(), bytes.end());
......@@ -47,7 +46,7 @@ bool server::send_payload(context, const_byte_span bytes) {
return true;
}
bool server::send_chunk(context, const_byte_span bytes) {
bool server::send_chunk(const_byte_span bytes) {
down_->begin_output();
auto& buf = down_->output_buffer();
auto size = bytes.size();
......@@ -69,10 +68,6 @@ bool server::send_end_of_chunks() {
return down_->end_output();
}
void server::fin(context) {
// nop
}
// -- stream_oriented::upper_layer implementation ------------------------------
error server::init(socket_manager* owner, stream_oriented::lower_layer* down,
......@@ -178,7 +173,7 @@ void server::write_response(status code, std::string_view content) {
}
bool server::invoke_upper_layer(const_byte_span payload) {
return up_->consume(context{}, hdr_, payload) >= 0;
return up_->consume(hdr_, payload) >= 0;
}
bool server::handle_header(std::string_view http) {
......
......@@ -167,12 +167,12 @@ operation multiplexer::mask_of(const socket_manager_ptr& mgr) {
// -- thread-safe signaling ----------------------------------------------------
void multiplexer::discard(const socket_manager_ptr& mgr) {
void multiplexer::dispose(const socket_manager_ptr& mgr) {
CAF_LOG_TRACE(CAF_ARG2("socket", mgr->handle().id));
if (std::this_thread::get_id() == tid_) {
do_discard(mgr);
do_dispose(mgr);
} else {
write_to_pipe(pollset_updater::code::discard_manager, mgr.get());
write_to_pipe(pollset_updater::code::dispose_manager, mgr.get());
}
}
......@@ -471,7 +471,7 @@ void multiplexer::do_shutdown() {
apply_updates();
}
void multiplexer::do_discard(const socket_manager_ptr& mgr) {
void multiplexer::do_dispose(const socket_manager_ptr& mgr) {
CAF_LOG_TRACE(CAF_ARG2("socket", mgr->handle().id));
mgr->handle_error(sec::disposed);
update_for(mgr.get()).events = 0;
......
......@@ -60,8 +60,8 @@ void pollset_updater::handle_read_event() {
case code::init_manager:
mpx_->do_init(as_mgr(ptr));
break;
case code::discard_manager:
mpx_->do_discard(as_mgr(ptr));
case code::dispose_manager:
mpx_->do_dispose(as_mgr(ptr));
break;
case code::shutdown_reading:
mpx_->do_shutdown_reading(as_mgr(ptr));
......
// 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/net/prometheus/server.hpp"
#include "caf/net/http/header.hpp"
#include "caf/net/http/lower_layer.hpp"
using namespace std::literals;
namespace caf::net::prometheus {
// -- server::scrape_state -----------------------------------------------------
std::string_view server::scrape_state::scrape() {
// Scrape system metrics at most once per second. TODO: make configurable.
if (auto now = std::chrono::steady_clock::now(); last_scrape + 1s <= now) {
last_scrape = now;
proc_importer.update();
}
return collector.collect_from(*registry);
}
// -- implementation of http::upper_layer --------------------------------------
bool server::prepare_send() {
return true;
}
bool server::done_sending() {
return true;
}
void server::abort(const error&) {
// nop
}
error server::init(socket_manager*, http::lower_layer* down, const settings&) {
down_ = down;
down_->request_messages();
return caf::none;
}
ptrdiff_t server::consume(const http::header& hdr, const_byte_span payload) {
if (hdr.path() != "/metrics") {
down_->send_response(http::status::not_found, "text/plain", "Not found.");
down_->close();
} else if (hdr.method() != http::method::get) {
down_->send_response(http::status::method_not_allowed, "text/plain",
"Method not allowed.");
down_->close();
} else if (!hdr.query().empty() || !hdr.fragment().empty()) {
down_->send_response(http::status::bad_request, "text/plain",
"No fragment or query allowed.");
down_->close();
} else {
auto str = state_->scrape();
down_->send_response(http::status::ok, "text/plain;version=0.0.4", str);
down_->close();
}
return static_cast<ptrdiff_t>(payload.size());
}
} // namespace caf::net::prometheus
......@@ -33,6 +33,49 @@ socket_manager_ptr socket_manager::make(multiplexer* mpx, socket handle,
return make_counted<socket_manager>(mpx, handle, std::move(handler));
}
namespace {
class disposer : public detail::atomic_ref_counted, public disposable_impl {
public:
disposer(multiplexer* mpx, socket_manager_ptr mgr)
: mpx_(mpx), mgr_(std::move(mgr)) {
// nop
}
void dispose() {
std::unique_lock guard{mtx_};
if (mpx_) {
mpx_->dispose(mgr_);
mpx_ = nullptr;
mgr_ = nullptr;
}
}
bool disposed() const noexcept {
std::unique_lock guard{mtx_};
return mpx_ == nullptr;
}
void ref_disposable() const noexcept {
ref();
}
void deref_disposable() const noexcept {
deref();
}
private:
mutable std::mutex mtx_;
multiplexer* mpx_;
socket_manager_ptr mgr_;
};
} // namespace
disposable socket_manager::make_disposer() {
return disposable{make_counted<disposer>(mpx_, this)};
}
// -- properties ---------------------------------------------------------------
actor_system& socket_manager::system() noexcept {
......
......@@ -67,12 +67,11 @@ public:
return true;
}
ptrdiff_t consume(net::http::context ctx,
const net::http::header& request_hdr,
ptrdiff_t consume(const net::http::header& request_hdr,
const_byte_span body) override {
hdr = request_hdr;
auto content = "Hello world!"sv;
down->send_response(ctx, net::http::status::ok, "text/plain",
down->send_response(net::http::status::ok, "text/plain",
as_bytes(make_span(content)));
payload.assign(body.begin(), body.end());
return static_cast<ptrdiff_t>(body.size());
......
// 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
// prometheuss://github.com/actor-framework/actor-framework/blob/master/LICENSE.
#define CAF_SUITE net.prometheus.server
#include "caf/net/prometheus/server.hpp"
#include "net-test.hpp"
#include "caf/net/http/server.hpp"
#include "caf/telemetry/metric_registry.hpp"
using namespace caf;
using namespace caf::net;
using namespace std::literals;
constexpr std::string_view request_str
= "GET /metrics HTTP/1.1\r\n"
"Host: localhost:8090\r\n"
"User-Agent: Prometheus/2.18.1\r\n"
"Accept: text/plain;version=0.0.4\r\n"
"Accept-Encoding: gzip\r\n"
"X-Prometheus-Scrape-Timeout-Seconds: 5.000000\r\n\r\n";
SCENARIO("the Prometheus server responds to requests with scrape results") {
GIVEN("a valid Prometheus GET request") {
WHEN("sending it to an prometheus server") {
THEN("the Prometheus server responds with metrics text data") {
telemetry::metric_registry registry;
auto fb = registry.counter_singleton("foo", "bar", "test metric");
auto bf = registry.counter_singleton("bar", "foo", "test metric");
fb->inc(3);
bf->inc(7);
auto prom_state = prometheus::server::scrape_state::make(&registry);
auto prom_serv = prometheus::server::make(prom_state);
auto http_serv = net::http::server::make(std::move(prom_serv));
auto serv = mock_stream_transport::make(std::move(http_serv));
CHECK_EQ(serv->init(settings{}), error{});
serv->push(request_str);
CHECK_EQ(serv->handle_input(),
static_cast<ptrdiff_t>(request_str.size()));
auto out = serv->output_as_str();
CHECK(out.find("foo_bar 3"sv) != std::string_view::npos);
CHECK(out.find("bar_foo 7"sv) != std::string_view::npos);
}
}
}
}
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