Commit 8c82c734 authored by Dominik Charousset's avatar Dominik Charousset

Add Prometheus broker for exporting sys metrics

parent 5e0fb483
......@@ -28,6 +28,7 @@ endfunction()
# -- add library target --------------------------------------------------------
add_library(libcaf_io_obj OBJECT ${CAF_IO_HEADERS}
src/detail/prometheus_broker.cpp
src/detail/socket_guard.cpp
src/io/abstract_broker.cpp
src/io/basp/header.cpp
......@@ -95,6 +96,7 @@ target_include_directories(caf-io-test PRIVATE "${CMAKE_CURRENT_SOURCE_DIR}/test
target_link_libraries(caf-io-test PRIVATE CAF::test)
caf_add_test_suites(caf-io-test
detail.prometheus_broker
io.basp.message_queue
io.basp_broker
io.broker
......
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright 2011-2020 Dominik Charousset *
* *
* Distributed under the terms and conditions of the BSD 3-Clause License or *
* (at your option) under the terms and conditions of the Boost Software *
* License 1.0. See accompanying files LICENSE and LICENSE_ALTERNATIVE. *
* *
* If you did not receive a copy of the license files, see *
* http://opensource.org/licenses/BSD-3-Clause and *
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#pragma once
#include <ctime>
#include <unordered_map>
#include <vector>
#include "caf/byte_buffer.hpp"
#include "caf/detail/io_export.hpp"
#include "caf/fwd.hpp"
#include "caf/io/broker.hpp"
#include "caf/telemetry/collector/prometheus.hpp"
namespace caf::detail {
/// Makes system metrics in the Prometheus format available via HTTP 1.1.
class CAF_IO_EXPORT prometheus_broker : public io::broker {
public:
explicit prometheus_broker(actor_config& cfg);
~prometheus_broker() override;
const char* name() const override;
static bool has_process_metrics() noexcept;
behavior make_behavior() override;
private:
void scrape();
std::unordered_map<io::connection_handle, byte_buffer> requests_;
telemetry::collector::prometheus collector_;
time_t last_scrape_ = 0;
telemetry::dbl_gauge* cpu_time_ = nullptr;
telemetry::int_gauge* mem_size_ = nullptr;
telemetry::int_gauge* virt_mem_size_ = nullptr;
};
} // namespace caf::detail
......@@ -158,6 +158,9 @@ public:
/// Writes `data` into the buffer for a given connection.
void write(connection_handle hdl, size_t bs, const void* buf);
/// Writes `buf` into the buffer for a given connection.
void write(connection_handle hdl, span<const byte> buf);
/// Sends the content of the buffer for a given connection.
void flush(connection_handle hdl);
......
......@@ -251,6 +251,13 @@ 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);
/// Returns a middleman using the default network backend.
static actor_system::module* make(actor_system&, detail::type_list<>);
......
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright 2011-2020 Dominik Charousset *
* *
* Distributed under the terms and conditions of the BSD 3-Clause License or *
* (at your option) under the terms and conditions of the Boost Software *
* License 1.0. See accompanying files LICENSE and LICENSE_ALTERNATIVE. *
* *
* If you did not receive a copy of the license files, see *
* http://opensource.org/licenses/BSD-3-Clause and *
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#include "caf/detail/prometheus_broker.hpp"
#include "caf/span.hpp"
#include "caf/string_algorithms.hpp"
#include "caf/string_view.hpp"
#include "caf/telemetry/dbl_gauge.hpp"
#include "caf/telemetry/int_gauge.hpp"
#ifdef CAF_MACOS
# include <mach/mach.h>
# include <mach/task.h>
# include <sys/resource.h>
# define HAS_PROCESS_METRICS
namespace {
std::atomic<std::remove_pointer_t<vm_size_t>> global_page_size;
std::pair<int64_t, int64_t> get_mem_usage() {
mach_task_basic_info info;
auto page_size = global_page_size.load();
if (page_size == 0) {
if (host_page_size(mach_host_self(), &page_size) == KERN_SUCCESS) {
global_page_size = page_size;
printf("page size = %d\n", (int) page_size);
} else {
puts("failed to get page size");
}
}
mach_msg_type_number_t count = MACH_TASK_BASIC_INFO_COUNT;
if (task_info(mach_task_self(), MACH_TASK_BASIC_INFO, (task_info_t) &info,
&count)
!= KERN_SUCCESS) {
puts("failed to get task info");
return {0, 0};
}
puts("got task info");
return {static_cast<int64_t>(info.resident_size),
static_cast<int64_t>(info.virtual_size)};
}
double get_cpu_time() {
task_thread_times_info info;
mach_msg_type_number_t count = TASK_THREAD_TIMES_INFO_COUNT;
if (task_info(mach_task_self(), TASK_THREAD_TIMES_INFO, (task_info_t) &info,
&count)
!= KERN_SUCCESS)
return 0;
// Round to milliseconds.
double result = 0;
result += info.user_time.seconds;
result += ceil(info.user_time.microseconds / 1000.0) / 1000.0;
result += info.system_time.seconds;
result += ceil(info.system_time.microseconds / 1000.0) / 1000.0;
return result;
}
} // namespace
#endif // CAF_MACOS
#ifdef CAF_LINUX
# include <sys/resource.h>
# define HAS_PROCESS_METRICS
std::pair<int64_t, int64_t> get_mem_usage() {
return {0, 0};
}
int64_t get_cpu_time() {
rusage usage;
if (getrusage(RUSAGE_SELF, &usage) != 0)
return 0;
return static_cast<int64_t>(usage.ru_utime.tv_sec + usage.ru_stime.tv_sec);
}
#endif // CAF_LINUX
namespace caf::detail {
namespace {
// Cap incoming HTTP requests.
constexpr size_t max_request_size = 512 * 1024;
// HTTP response for requests that exceed the size limit.
constexpr string_view request_too_large
= "HTTP/1.1 413 Request Entity Too Large\r\n"
"Connection: Closed\r\n\r\n";
// HTTP response for requests that aren't "GET /metrics HTTP/1.1".
constexpr string_view request_not_supported = "HTTP/1.1 501 Not Implemented\r\n"
"Connection: Closed\r\n\r\n";
// HTTP header when sending a payload.
constexpr string_view request_ok = "HTTP/1.1 200 OK\r\n"
"Content-Type: text/plain\r\n"
"Connection: Closed\r\n\r\n";
} // namespace
prometheus_broker::prometheus_broker(actor_config& cfg) : io::broker(cfg) {
using telemetry::dbl_gauge;
using telemetry::int_gauge;
auto& registry = system().telemetry();
#ifdef HAS_PROCESS_METRICS
cpu_time_ = registry.singleton<dbl_gauge>(
"process", "cpu", "Total user and system CPU time spent.", "seconds", true);
mem_size_ = registry.singleton<int_gauge>(
"process", "resident_memory", " Resident memory size.", "bytes");
virt_mem_size_ = registry.singleton<int_gauge>(
"process", "virtual_memory", " Virtual memory size.", "bytes");
#endif // HAS_PROCESS_METRICS
}
prometheus_broker::~prometheus_broker() {
// nop
}
const char* prometheus_broker::name() const {
return "prometheus-broker";
}
bool prometheus_broker::has_process_metrics() noexcept {
#ifdef HAS_PROCESS_METRICS
return true;
#else // HAS_PROCESS_METRICS
return false;
#endif // HAS_PROCESS_METRICS
}
behavior prometheus_broker::make_behavior() {
return {
[=](const io::new_data_msg& msg) {
auto& req = requests_[msg.handle];
if (req.size() + msg.buf.size() > max_request_size) {
write(msg.handle, as_bytes(make_span(request_too_large)));
close(msg.handle);
return;
}
req.insert(req.end(), msg.buf.begin(), msg.buf.end());
auto req_str = string_view{reinterpret_cast<char*>(req.data()),
req.size()};
// Stop here if the header isn't complete yet.
if (!ends_with(req_str, "\r\n\r\n"))
return;
// We only check whether it's a GET request for /metrics for HTTP 1.x.
// Everything else, we ignore for now.
if (!starts_with(req_str, "GET /metrics HTTP/1.")) {
write(msg.handle, as_bytes(make_span(request_not_supported)));
close(msg.handle);
return;
}
// Collect metrics, ship response, and close.
scrape();
auto hdr = as_bytes(make_span(request_ok));
auto text = collector_.collect_from(system().telemetry());
auto payload = as_bytes(make_span(text));
auto& dst = wr_buf(msg.handle);
dst.insert(dst.end(), hdr.begin(), hdr.end());
dst.insert(dst.end(), payload.begin(), payload.end());
close(msg.handle);
},
[=](const io::new_connection_msg& msg) {
// Pre-allocate buffer for maximum request size.
auto& req = requests_[msg.handle];
req.reserve(512 * 1024);
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);
},
[=](const io::acceptor_closed_msg&) {
// Shoud not happen.
quit(sec::socket_operation_failed);
},
};
}
void prometheus_broker::scrape() {
#ifdef HAS_PROCESS_METRICS
// Collect system metrics at most once per second.
auto now = time(NULL);
if (last_scrape_ >= now)
return;
last_scrape_ = now;
cpu_time_->value(get_cpu_time());
auto [res, virt] = get_mem_usage();
mem_size_->value(res);
virt_mem_size_->value(virt);
#endif // HAS_PROCESS_METRICS
}
} // namespace caf::detail
......@@ -22,6 +22,7 @@
#include "caf/make_counted.hpp"
#include "caf/none.hpp"
#include "caf/scheduler/abstract_coordinator.hpp"
#include "caf/span.hpp"
#include "caf/io/broker.hpp"
#include "caf/io/middleman.hpp"
......@@ -104,6 +105,10 @@ void abstract_broker::write(connection_handle hdl, size_t bs, const void* buf) {
out.insert(out.end(), first, last);
}
void abstract_broker::write(connection_handle hdl, span<const byte> buf) {
write(hdl, buf.size(), buf.data());
}
void abstract_broker::flush(connection_handle hdl) {
auto x = by_id(hdl);
if (x)
......
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright 2011-2020 Dominik Charousset *
* *
* Distributed under the terms and conditions of the BSD 3-Clause License or *
* (at your option) under the terms and conditions of the Boost Software *
* License 1.0. See accompanying files LICENSE and LICENSE_ALTERNATIVE. *
* *
* If you did not receive a copy of the license files, see *
* http://opensource.org/licenses/BSD-3-Clause and *
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#define CAF_SUITE detail.prometheus_broker
#include "caf/detail/prometheus_broker.hpp"
#include "caf/test/io_dsl.hpp"
using namespace caf;
using namespace caf::io;
namespace {
struct fixture : test_node_fixture<> {
caf::actor aut;
fixture() {
using detail::prometheus_broker;
actor_config cfg{&sys.middleman().backend()};
aut = sys.spawn_impl<prometheus_broker, spawn_options::no_flags>(cfg);
run();
// assign the acceptor handle to the AUT
auto ptr = static_cast<abstract_broker*>(actor_cast<abstract_actor*>(aut));
ptr->add_doorman(mpx.new_doorman(acceptor, 1u));
// "open" a new connection to our server
mpx.add_pending_connect(acceptor, connection);
mpx.accept_connection(acceptor);
}
~fixture() {
anon_send(aut, exit_reason::user_shutdown);
run();
}
accept_handle acceptor = accept_handle::from_int(1);
connection_handle connection = connection_handle::from_int(1);
};
bool contains(string_view str, string_view what) {
return str.find(what) != string_view::npos;
}
} // 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));
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(contains(response, "\ncaf_running_actors 2 "));
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 "));
}
}
CAF_TEST_FIXTURE_SCOPE_END()
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