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

Add process metrics importer

Move the logic for importing process metrics from the Prometheus broker
to public API. This allows users to get access to the process metrics
without having to load the HTTP server for Prometheus scraping.
parent 642612a3
......@@ -5,6 +5,12 @@ is based on [Keep a Changelog](https://keepachangelog.com).
## [Unreleased]
### Added
- The new class `caf::telemetry::importer::process` allows users to get access
to process metrics even when not configuring CAF to export metrics to
Prometheus via HTTP.
## Fixed
- Printing a `config_value` that contains a zero duration `timespan` now
......
......@@ -204,6 +204,7 @@ caf_add_component(
src/telemetry/metric.cpp
src/telemetry/metric_family.cpp
src/telemetry/metric_registry.cpp
src/telemetry/importer/process.cpp
src/term.cpp
src/thread_hook.cpp
src/timestamp.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/core_export.hpp"
#include "caf/fwd.hpp"
namespace caf::telemetry::importer {
/// Imports CPU and memory metrics for the current process. On supported
/// platforms, this importer adds the metrics `process.resident_memory`
/// (resident memory size), `process.virtual_memory` (virtual memory size) and
/// `process.cpu` (total user and system CPU time spent).
///
/// @note CAF adds this importer automatically when configuring export to
/// Prometheus via HTTP.
class CAF_CORE_EXPORT process {
public:
explicit process(metric_registry& reg);
/// Returns whether the scraper supports the host platform.
static bool platform_supported() noexcept;
/// Updates process metrics.
/// @note has no effect if `platform_supported()` returns `false`.
void update();
private:
telemetry::int_gauge* rss_ = nullptr;
telemetry::int_gauge* vms_ = nullptr;
telemetry::dbl_gauge* cpu_ = nullptr;
};
} // namespace caf::telemetry::importer
// 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/telemetry/importer/process.hpp"
#include "caf/logger.hpp"
#include "caf/telemetry/gauge.hpp"
#include "caf/telemetry/metric_registry.hpp"
// -- detect supported platforms -----------------------------------------------
#if defined(CAF_MACOS) || defined(CAF_LINUX)
# define CAF_HAS_PROCESS_METRICS
#endif
// -- gauge setup and boilerplate ----------------------------------------------
#ifdef CAF_HAS_PROCESS_METRICS
namespace {
struct sys_stats {
int64_t rss;
int64_t vms;
double cpu_time;
};
sys_stats read_sys_stats();
static constexpr bool platform_supported_v = true;
template <class CpuPtr, class RssPtr, class VmsPtr>
void sys_stats_init(caf::telemetry::metric_registry& reg, RssPtr& rss,
VmsPtr& vms, CpuPtr& cpu) {
rss = reg.gauge_singleton("process", "resident_memory",
"Resident memory size.", "bytes");
vms = reg.gauge_singleton("process", "virtual_memory", "Virtual memory size.",
"bytes");
cpu = reg.gauge_singleton<double>("process", "cpu",
"Total user and system CPU time spent.",
"seconds", true);
}
void update_impl(caf::telemetry::int_gauge* rss_gauge,
caf::telemetry::int_gauge* vms_gauge,
caf::telemetry::dbl_gauge* cpu_gauge) {
auto [rss, vmsize, cpu] = read_sys_stats();
rss_gauge->value(rss);
vms_gauge->value(vmsize);
cpu_gauge->value(cpu);
}
} // namespace
#else // CAF_HAS_PROCESS_METRICS
namespace {
static constexpr bool platform_supported_v = false;
template <class... Ts>
void sys_stats_init(Ts&&...) {
// nop
}
template <class... Ts>
void update_impl(Ts&&...) {
// nop
}
} // namespace
#endif // CAF_HAS_PROCESS_METRICS
// -- macOS-specific scraping logic --------------------------------------------
#ifdef CAF_MACOS
# include <mach/mach.h>
# include <mach/task.h>
# include <sys/resource.h>
namespace {
sys_stats read_sys_stats() {
sys_stats result{0, 0, 0};
// Fetch memory usage.
{
mach_task_basic_info info;
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) {
result.rss = info.resident_size;
result.vms = info.virtual_size;
}
}
// Fetch 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) {
// Round to milliseconds.
result.cpu_time += info.user_time.seconds;
result.cpu_time += ceil(info.user_time.microseconds / 1000.0) / 1000.0;
result.cpu_time += info.system_time.seconds;
result.cpu_time += ceil(info.system_time.microseconds / 1000.0) / 1000.0;
}
}
return result;
}
} // namespace
#endif // CAF_MACOS
// -- Linux-specific scraping logic --------------------------------------------
#ifdef CAF_LINUX
# include <cstdio>
# include <unistd.h>
namespace {
std::atomic<long> global_ticks_per_second;
std::atomic<long> global_page_size;
bool load_system_setting(std::atomic<long>& cache_var, long& var, int name,
[[maybe_unused]] const char* pretty_name) {
var = cache_var.load();
switch (var) {
case -1:
return false;
case 0:
var = sysconf(name);
if (var <= 0) {
CAF_LOG_ERROR("failed to read" << pretty_name << "from sysconf");
var = -1;
cache_var = var;
return false;
}
cache_var = var;
return true;
default:
return true;
}
}
# define TRY_LOAD(varname, confname) \
load_system_setting(global_##varname, varname, confname, #confname)
sys_stats read_sys_stats() {
sys_stats result{0, 0, 0};
long ticks_per_second = 0;
long page_size = 0;
if (!TRY_LOAD(ticks_per_second, _SC_CLK_TCK)
|| !TRY_LOAD(page_size, _SC_PAGE_SIZE))
return result;
if (auto f = fopen("/proc/self/stat", "r")) {
long unsigned utime_ticks = 0;
long unsigned stime_ticks = 0;
long unsigned vmsize_bytes = 0;
long rss_pages = 0;
auto rd = fscanf(f,
"%*d " // 1. PID
"%*s " // 2. Executable
"%*c " // 3. State
"%*d " // 4. Parent PID
"%*d " // 5. Process group ID
"%*d " // 6. Session ID
"%*d " // 7. Controlling terminal
"%*d " // 8. Foreground process group ID
"%*u " // 9. Flags
"%*u " // 10. Number of minor faults
"%*u " // 11. Number of minor faults of waited-for children
"%*u " // 12. Number of major faults
"%*u " // 13. Number of major faults of waited-for children
"%lu " // 14. CPU user time in ticks
"%lu " // 15. CPU kernel time in ticks
"%*d " // 16. CPU user time of waited-for children
"%*d " // 17. CPU kernel time of waited-for children
"%*d " // 18. Priority
"%*d " // 19. Nice value
"%*d " // 20. Num threads
"%*d " // 21. Obsolete since 2.6
"%*u " // 22. Time the process started after system boot
"%lu " // 23. Virtual memory size in bytes
"%ld", // 24. Resident set size in pages
&utime_ticks, &stime_ticks, &vmsize_bytes, &rss_pages);
fclose(f);
if (rd != 4) {
CAF_LOG_ERROR("failed to read content of /proc/self/stat");
global_ticks_per_second = -1;
global_page_size = -1;
return result;
}
result.rss = static_cast<int64_t>(rss_pages) * page_size;
result.vms = static_cast<int64_t>(vmsize_bytes);
result.cpu_time = utime_ticks;
result.cpu_time += stime_ticks;
result.cpu_time /= ticks_per_second;
}
return result;
}
} // namespace
#endif // CAF_LINUX
namespace caf::telemetry::importer {
process::process(metric_registry& reg) {
sys_stats_init(reg, rss_, vms_, cpu_);
}
bool process::platform_supported() noexcept {
return platform_supported_v;
}
void process::update() {
update_impl(rss_, vms_, cpu_);
}
} // namespace caf::telemetry::importer
......@@ -13,6 +13,7 @@
#include "caf/fwd.hpp"
#include "caf/io/broker.hpp"
#include "caf/telemetry/collector/prometheus.hpp"
#include "caf/telemetry/importer/process.hpp"
namespace caf::detail {
......@@ -27,7 +28,9 @@ public:
const char* name() const override;
static bool has_process_metrics() noexcept;
static bool has_process_metrics() noexcept {
return telemetry::importer::process::platform_supported();
}
behavior make_behavior() override;
......@@ -37,9 +40,7 @@ private:
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;
telemetry::importer::process proc_importer_;
};
} // namespace caf::detail
......@@ -10,146 +10,6 @@
#include "caf/telemetry/dbl_gauge.hpp"
#include "caf/telemetry/int_gauge.hpp"
namespace {
struct [[maybe_unused]] sys_stats {
int64_t rss;
int64_t vmsize;
double cpu_time;
};
} // namespace
#ifdef CAF_MACOS
# include <mach/mach.h>
# include <mach/task.h>
# include <sys/resource.h>
# define HAS_PROCESS_METRICS
namespace {
sys_stats read_sys_stats() {
sys_stats result{0, 0, 0};
// Fetch memory usage.
{
mach_task_basic_info info;
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) {
result.rss = info.resident_size;
result.vmsize = info.virtual_size;
}
}
// Fetch 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) {
// Round to milliseconds.
result.cpu_time += info.user_time.seconds;
result.cpu_time += ceil(info.user_time.microseconds / 1000.0) / 1000.0;
result.cpu_time += info.system_time.seconds;
result.cpu_time += ceil(info.system_time.microseconds / 1000.0) / 1000.0;
}
}
return result;
}
} // namespace
#endif // CAF_MACOS
#ifdef CAF_LINUX
# include <cstdio>
# include <unistd.h>
# define HAS_PROCESS_METRICS
namespace {
std::atomic<long> global_ticks_per_second;
std::atomic<long> global_page_size;
bool load_system_setting(std::atomic<long>& cache_var, long& var, int name,
[[maybe_unused]] const char* pretty_name) {
var = cache_var.load();
switch (var) {
case -1:
return false;
case 0:
var = sysconf(name);
if (var <= 0) {
CAF_LOG_ERROR("failed to read" << pretty_name << "from sysconf");
var = -1;
cache_var = var;
return false;
}
cache_var = var;
return true;
default:
return true;
}
}
# define TRY_LOAD(varname, confname) \
load_system_setting(global_##varname, varname, confname, #confname)
sys_stats read_sys_stats() {
sys_stats result{0, 0, 0};
long ticks_per_second = 0;
long page_size = 0;
if (!TRY_LOAD(ticks_per_second, _SC_CLK_TCK)
|| !TRY_LOAD(page_size, _SC_PAGE_SIZE))
return result;
if (auto f = fopen("/proc/self/stat", "r")) {
long unsigned utime_ticks = 0;
long unsigned stime_ticks = 0;
long unsigned vmsize_bytes = 0;
long rss_pages = 0;
auto rd = fscanf(f,
"%*d " // 1. PID
"%*s " // 2. Executable
"%*c " // 3. State
"%*d " // 4. Parent PID
"%*d " // 5. Process group ID
"%*d " // 6. Session ID
"%*d " // 7. Controlling terminal
"%*d " // 8. Foreground process group ID
"%*u " // 9. Flags
"%*u " // 10. Number of minor faults
"%*u " // 11. Number of minor faults of waited-for children
"%*u " // 12. Number of major faults
"%*u " // 13. Number of major faults of waited-for children
"%lu " // 14. CPU user time in ticks
"%lu " // 15. CPU kernel time in ticks
"%*d " // 16. CPU user time of waited-for children
"%*d " // 17. CPU kernel time of waited-for children
"%*d " // 18. Priority
"%*d " // 19. Nice value
"%*d " // 20. Num threads
"%*d " // 21. Obsolete since 2.6
"%*u " // 22. Time the process started after system boot
"%lu " // 23. Virtual memory size in bytes
"%ld", // 24. Resident set size in pages
&utime_ticks, &stime_ticks, &vmsize_bytes, &rss_pages);
fclose(f);
if (rd != 4) {
CAF_LOG_ERROR("failed to read content of /proc/self/stat");
global_ticks_per_second = -1;
global_page_size = -1;
return result;
}
result.rss = static_cast<int64_t>(rss_pages) * page_size;
result.vmsize = static_cast<int64_t>(vmsize_bytes);
result.cpu_time = utime_ticks;
result.cpu_time += stime_ticks;
result.cpu_time /= ticks_per_second;
}
return result;
}
} // namespace
#endif // CAF_LINUX
namespace caf::detail {
namespace {
......@@ -173,18 +33,9 @@ constexpr string_view request_ok = "HTTP/1.1 200 OK\r\n"
} // namespace
prometheus_broker::prometheus_broker(actor_config& cfg) : io::broker(cfg) {
#ifdef HAS_PROCESS_METRICS
using telemetry::dbl_gauge;
using telemetry::int_gauge;
auto& reg = system().metrics();
cpu_time_ = reg.gauge_singleton<double>(
"process", "cpu", "Total user and system CPU time spent.", "seconds", true);
mem_size_ = reg.gauge_singleton("process", "resident_memory",
"Resident memory size.", "bytes");
virt_mem_size_ = reg.gauge_singleton("process", "virtual_memory",
"Virtual memory size.", "bytes");
#endif // HAS_PROCESS_METRICS
prometheus_broker::prometheus_broker(actor_config& cfg)
: io::broker(cfg), proc_importer_(system().metrics()) {
// nop
}
prometheus_broker::prometheus_broker(actor_config& cfg, io::doorman_ptr ptr)
......@@ -200,14 +51,6 @@ const char* prometheus_broker::name() const {
return "caf.system.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) {
......@@ -267,17 +110,12 @@ behavior prometheus_broker::make_behavior() {
}
void prometheus_broker::scrape() {
#ifdef HAS_PROCESS_METRICS
// Collect system metrics at most once per second.
// Scrape system metrics at most once per second.
auto now = time(NULL);
if (last_scrape_ >= now)
return;
last_scrape_ = now;
auto [rss, vmsize, cpu_time] = read_sys_stats();
mem_size_->value(rss);
virt_mem_size_->value(vmsize);
cpu_time_->value(cpu_time);
#endif // HAS_PROCESS_METRICS
if (last_scrape_ < now) {
last_scrape_ = now;
proc_importer_.update();
}
}
} // namespace caf::detail
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