Unverified Commit 16b3f362 authored by Dominik Charousset's avatar Dominik Charousset Committed by GitHub

Merge pull request #1255

Fix shutdown of the Prometheus background task
parents e5b93fd1 4a8dd72c
......@@ -35,6 +35,7 @@ is based on [Keep a Changelog](https://keepachangelog.com).
the ambiguity, `result<T>` now accepts any type that allows constructing a `T`
internally without requiring a type conversion to `T` as an argument (#1245).
- Fix configuration parameter lookup for the `work-stealing` scheduler policy.
- Applications that expose metrics to Prometheus properly terminate now.
## [0.18.2] - 2021-03-26
......
......@@ -110,6 +110,7 @@ caf_add_component(
src/detail/group_tunnel.cpp
src/detail/invoke_result_visitor.cpp
src/detail/json.cpp
src/detail/latch.cpp
src/detail/local_group_module.cpp
src/detail/message_builder_element.cpp
src/detail/message_data.cpp
......@@ -251,6 +252,7 @@ caf_add_component(
detail.group_tunnel
detail.ieee_754
detail.json
detail.latch
detail.limited_vector
detail.local_group_module
detail.meta_object
......
// 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 <condition_variable>
#include <cstdint>
#include <mutex>
#include "caf/detail/core_export.hpp"
namespace caf::detail {
// Drop-in replacement for C++20's std::latch.
class CAF_CORE_EXPORT latch {
public:
explicit latch(ptrdiff_t value) : count_(value) {
// nop
}
latch(const latch&) = delete;
latch& operator=(const latch&) = delete;
void count_down_and_wait();
void wait();
void count_down();
bool is_ready() const noexcept;
private:
mutable std::mutex mtx_;
std::condition_variable cv_;
ptrdiff_t count_;
};
} // 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.
#include "caf/detail/latch.hpp"
namespace caf::detail {
namespace {
using guard_type = std::unique_lock<std::mutex>;
} // namespace
void latch::count_down_and_wait() {
guard_type guard{mtx_};
if (--count_ == 0) {
cv_.notify_all();
} else {
do {
cv_.wait(guard);
} while (count_ > 0);
}
}
void latch::wait() {
guard_type guard{mtx_};
while (count_ > 0)
cv_.wait(guard);
}
void latch::count_down() {
guard_type guard{mtx_};
if (--count_ == 0)
cv_.notify_all();
}
bool latch::is_ready() const noexcept {
guard_type guard{mtx_};
return count_ == 0;
}
} // 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.
#define CAF_SUITE detail.latch
#include "caf/detail/latch.hpp"
#include "core-test.hpp"
using namespace caf;
SCENARIO("latches synchronize threads") {
GIVEN("a latch and three threads") {
detail::latch sync{2};
std::vector<std::thread> threads;
WHEN("synchronizing the threads via the latch") {
THEN("wait() blocks until all threads counted down the latch") {
threads.emplace_back([&sync] { sync.count_down(); });
threads.emplace_back([&sync] { sync.count_down_and_wait(); });
threads.emplace_back([&sync] { sync.wait(); });
sync.wait();
CHECK(sync.is_ready());
}
}
for (auto& t : threads)
t.join();
}
}
......@@ -32,7 +32,7 @@ namespace caf::io {
/// Manages brokers and network backends.
class CAF_IO_EXPORT middleman : public actor_system::networking_module {
public:
friend class ::caf::actor_system;
friend class actor_system;
/// Metrics that the middleman collects by default.
struct metric_singletons_t {
......@@ -52,7 +52,6 @@ public:
/// 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>;
......@@ -309,6 +308,11 @@ public:
/// @private
metric_singletons_t metric_singletons;
/// @private
uint16_t prometheus_scraping_port() const noexcept {
return prometheus_scraping_port_;
}
protected:
middleman(actor_system& sys);
......@@ -349,12 +353,6 @@ 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>
remote_spawn_impl(const node_id& nid, std::string& name, message& args,
std::set<std::string> s, timespan timeout);
......@@ -390,6 +388,10 @@ private:
/// Manages groups that run on a different node in the network.
detail::remote_group_module_ptr remote_groups_;
/// Stores the port where the Prometheus scraper is listening at (0 if no
/// scraper is running in the background).
uint16_t prometheus_scraping_port_ = 0;
};
} // namespace caf::io
......@@ -20,6 +20,7 @@
#include "caf/defaults.hpp"
#include "caf/detail/get_mac_addresses.hpp"
#include "caf/detail/get_root_uuid.hpp"
#include "caf/detail/latch.hpp"
#include "caf/detail/prometheus_broker.hpp"
#include "caf/detail/ripemd_160.hpp"
#include "caf/detail/safe_equal.hpp"
......@@ -113,7 +114,7 @@ public:
// nop
}
bool start(const config_value::dictionary& cfg) override {
expected<uint16_t> start(const config_value::dictionary& cfg) {
// Read port, address and reuse flag from the config.
uint16_t port = 0;
if (auto cfg_port = get_as<uint16_t>(cfg, "port")) {
......@@ -125,41 +126,54 @@ public:
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;
}
return start(port, addr, get_or(cfg, "reuse", 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))
if (auto maybe_dptr = mpx_.new_tcp_doorman(port, in, reuse)) {
dptr = std::move(*maybe_dptr);
else
return std::move(maybe_dptr.error());
} else {
auto& err = maybe_dptr.error();
CAF_LOG_ERROR("failed to expose Prometheus metrics:" << err);
return std::move(err);
}
auto actual_port = dptr->port();
// Spawn the actor and store its handle in background_brokers_.
using impl = detail::prometheus_broker;
mpx_supervisor_ = mpx_.make_supervisor();
actor_config cfg{&mpx_};
broker_ = mpx_.system().spawn_impl<impl, hidden>(cfg, std::move(dptr));
thread_ = mpx_.system().launch_thread("caf.io.prom",
[this] { mpx_.run(); });
detail::latch sync{1};
auto run_mpx = [this, sync_ptr{&sync}] {
CAF_LOG_TRACE("");
mpx_.thread_id(std::this_thread::get_id());
sync_ptr->count_down();
mpx_.run();
};
thread_ = mpx_.system().launch_thread("caf.io.prom", run_mpx);
sync.wait();
CAF_LOG_INFO("expose Prometheus metrics at port" << actual_port);
return actual_port;
}
~prometheus_scraping() {
if (broker_) {
anon_send_exit(broker_, exit_reason::user_shutdown);
if (mpx_supervisor_) {
mpx_.dispatch([=] {
auto ptr = static_cast<broker*>(actor_cast<abstract_actor*>(broker_));
if (!ptr->getf(abstract_actor::is_terminated_flag)) {
ptr->context(&mpx_);
ptr->quit();
ptr->finalize();
}
});
mpx_supervisor_.reset();
thread_.join();
}
}
private:
network::default_multiplexer mpx_;
network::multiplexer::supervisor_ptr mpx_supervisor_;
actor broker_;
std::thread thread_;
};
......@@ -411,8 +425,11 @@ void middleman::start() {
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))
if (auto port = ptr->start(*prom)) {
CAF_ASSERT(*port != 0);
prometheus_scraping_port_ = *port;
background_tasks_.emplace_back(std::move(ptr));
}
}
// Launch backend.
if (!get_or(config(), "caf.middleman.manual-multiplexing", false))
......@@ -422,23 +439,15 @@ void middleman::start() {
// thread instead. Other backends can set `middleman_detach_multiplexer` to
// false to suppress creation of the supervisor.
if (backend_supervisor_ != nullptr) {
std::atomic<bool> init_done{false};
std::mutex mtx;
std::condition_variable cv;
auto run_backend = [this, &mtx, &cv, &init_done] {
detail::latch sync{1};
auto run_backend = [this, sync_ptr{&sync}] {
CAF_LOG_TRACE("");
{
std::unique_lock<std::mutex> guard{mtx};
backend().thread_id(std::this_thread::get_id());
init_done = true;
cv.notify_one();
}
backend().thread_id(std::this_thread::get_id());
sync_ptr->count_down();
backend().run();
};
thread_ = system().launch_thread("caf.io.mpx", run_backend);
std::unique_lock<std::mutex> guard{mtx};
while (init_done == false)
cv.wait(guard);
sync.wait();
}
// Spawn utility actors.
auto basp = named_broker<basp_broker>("BASP");
......
......@@ -6,7 +6,10 @@
#include "caf/detail/prometheus_broker.hpp"
#include "caf/test/io_dsl.hpp"
#include "io-test.hpp"
#include "caf/io/network/default_multiplexer.hpp"
#include "caf/policy/tcp.hpp"
using namespace caf;
using namespace caf::io;
......@@ -42,29 +45,31 @@ bool contains(string_view str, string_view what) {
return str.find(what) != string_view::npos;
}
constexpr string_view http_request
= "GET /metrics HTTP/1.1\r\n"
"Host: localhost:8090\r\n"
"User-Agent: Prometheus/2.18.1\r\n"
"Accept: application/openmetrics-text; "
"version=0.0.1,text/plain;version=0.0.4;q=0.5,*/*;q=0.1\r\n"
"Accept-Encoding: gzip\r\n"
"X-Prometheus-Scrape-Timeout-Seconds: 5.000000\r\n\r\n";
constexpr string_view http_ok_header = "HTTP/1.1 200 OK\r\n"
"Content-Type: text/plain\r\n"
"Connection: Closed\r\n\r\n";
} // namespace
CAF_TEST_FIXTURE_SCOPE(prometheus_broker_tests, fixture)
CAF_TEST(the prometheus broker responds to HTTP get requests) {
string_view request
= "GET /metrics HTTP/1.1\r\n"
"Host: localhost:8090\r\n"
"User-Agent: Prometheus/2.18.1\r\n"
"Accept: application/openmetrics-text; "
"version=0.0.1,text/plain;version=0.0.4;q=0.5,*/*;q=0.1\r\n"
"Accept-Encoding: gzip\r\n"
"X-Prometheus-Scrape-Timeout-Seconds: 5.000000\r\n\r\n";
auto bytes = as_bytes(make_span(request));
auto bytes = as_bytes(make_span(http_request));
mpx.virtual_send(connection, byte_buffer{bytes.begin(), bytes.end()});
run();
auto& response_buf = mpx.output_buffer(connection);
string_view response{reinterpret_cast<char*>(response_buf.data()),
response_buf.size()};
string_view ok_header = "HTTP/1.1 200 OK\r\n"
"Content-Type: text/plain\r\n"
"Connection: Closed\r\n\r\n";
CAF_CHECK(starts_with(response, ok_header));
CAF_CHECK(starts_with(response, http_ok_header));
CAF_CHECK(contains(response, "\ncaf_system_running_actors 2 "));
if (detail::prometheus_broker::has_process_metrics()) {
CAF_CHECK(contains(response, "\nprocess_cpu_seconds_total "));
......@@ -74,3 +79,66 @@ CAF_TEST(the prometheus broker responds to HTTP get requests) {
}
CAF_TEST_FIXTURE_SCOPE_END()
namespace {
static constexpr size_t chunk_size = 1024;
using io::network::native_socket;
std::vector<char> read_all(string_view query, native_socket fd) {
while (!query.empty()) {
size_t written = 0;
policy::tcp::write_some(written, fd, query.data(), query.size());
query.remove_prefix(written);
}
std::vector<char> buf;
char chunk[chunk_size];
memset(chunk, 0, chunk_size);
for (;;) {
size_t received = 0;
auto state = policy::tcp::read_some(received, fd, chunk, chunk_size);
if (received > 0)
buf.insert(buf.end(), chunk, chunk + received);
if (state == io::network::rw_state::failure) {
io::network::close_socket(fd);
return buf;
}
}
}
std::vector<char> read_all(string_view query, const std::string& host,
uint16_t port) {
if (auto fd = io::network::new_tcp_connection(host, port))
return read_all(query, *fd);
else
FAIL("new_tcp_connection: " << to_string(fd.error()));
}
} // namespace
SCENARIO("setting caf.middleman.prometheus-http.port exports metrics to HTTP") {
GIVEN("a config with an entry for caf.middleman.prometheus-http.port") {
actor_system_config cfg;
cfg.load<io::middleman>();
cfg.set("caf.scheduler.max-threads", 2);
cfg.set("caf.middleman.prometheus-http.port", 0);
WHEN("starting an actor system using the config") {
actor_system sys{cfg};
THEN("the middleman creates a background task for HTTP requests") {
auto scraping_port = sys.middleman().prometheus_scraping_port();
REQUIRE_NE(scraping_port, 0);
auto response_buf = read_all(http_request, "localhost", scraping_port);
string_view response{reinterpret_cast<char*>(response_buf.data()),
response_buf.size()};
CAF_CHECK(starts_with(response, http_ok_header));
CAF_CHECK(contains(response, "\ncaf_system_running_actors "));
if (detail::prometheus_broker::has_process_metrics()) {
CAF_CHECK(contains(response, "\nprocess_cpu_seconds_total "));
CAF_CHECK(contains(response, "\nprocess_resident_memory_bytes "));
CAF_CHECK(contains(response, "\nprocess_virtual_memory_bytes "));
}
}
}
}
}
#include "caf/test/bdd_dsl.hpp"
#include "caf/test/io_dsl.hpp"
using calculator = caf::typed_actor<
......
......@@ -47,6 +47,8 @@
#define MESSAGE(what) CAF_MESSAGE(what)
#define FAIL(what) CAF_FAIL(what)
#define BEGIN_FIXTURE_SCOPE(fixture_class) \
CAF_TEST_FIXTURE_SCOPE(CAF_UNIFYN(tests), fixture_class)
......
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