Unverified Commit 3615a07f authored by Noir's avatar Noir Committed by GitHub

Merge pull request #1278

Add process metrics importer
parents cff048a4 de34dffe
...@@ -5,6 +5,12 @@ is based on [Keep a Changelog](https://keepachangelog.com). ...@@ -5,6 +5,12 @@ is based on [Keep a Changelog](https://keepachangelog.com).
## [Unreleased] ## [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 ## Fixed
- Printing a `config_value` that contains a zero duration `timespan` now - Printing a `config_value` that contains a zero duration `timespan` now
......
...@@ -204,6 +204,7 @@ caf_add_component( ...@@ -204,6 +204,7 @@ caf_add_component(
src/telemetry/metric.cpp src/telemetry/metric.cpp
src/telemetry/metric_family.cpp src/telemetry/metric_family.cpp
src/telemetry/metric_registry.cpp src/telemetry/metric_registry.cpp
src/telemetry/importer/process.cpp
src/term.cpp src/term.cpp
src/thread_hook.cpp src/thread_hook.cpp
src/timestamp.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,
reinterpret_cast<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,
reinterpret_cast<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 @@ ...@@ -13,6 +13,7 @@
#include "caf/fwd.hpp" #include "caf/fwd.hpp"
#include "caf/io/broker.hpp" #include "caf/io/broker.hpp"
#include "caf/telemetry/collector/prometheus.hpp" #include "caf/telemetry/collector/prometheus.hpp"
#include "caf/telemetry/importer/process.hpp"
namespace caf::detail { namespace caf::detail {
...@@ -27,7 +28,9 @@ public: ...@@ -27,7 +28,9 @@ public:
const char* name() const override; 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; behavior make_behavior() override;
...@@ -37,9 +40,7 @@ private: ...@@ -37,9 +40,7 @@ private:
std::unordered_map<io::connection_handle, byte_buffer> requests_; std::unordered_map<io::connection_handle, byte_buffer> requests_;
telemetry::collector::prometheus collector_; telemetry::collector::prometheus collector_;
time_t last_scrape_ = 0; time_t last_scrape_ = 0;
telemetry::dbl_gauge* cpu_time_ = nullptr; telemetry::importer::process proc_importer_;
telemetry::int_gauge* mem_size_ = nullptr;
telemetry::int_gauge* virt_mem_size_ = nullptr;
}; };
} // namespace caf::detail } // namespace caf::detail
...@@ -10,146 +10,6 @@ ...@@ -10,146 +10,6 @@
#include "caf/telemetry/dbl_gauge.hpp" #include "caf/telemetry/dbl_gauge.hpp"
#include "caf/telemetry/int_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 caf::detail {
namespace { namespace {
...@@ -173,18 +33,9 @@ constexpr string_view request_ok = "HTTP/1.1 200 OK\r\n" ...@@ -173,18 +33,9 @@ constexpr string_view request_ok = "HTTP/1.1 200 OK\r\n"
} // namespace } // namespace
prometheus_broker::prometheus_broker(actor_config& cfg) : io::broker(cfg) { prometheus_broker::prometheus_broker(actor_config& cfg)
#ifdef HAS_PROCESS_METRICS : io::broker(cfg), proc_importer_(system().metrics()) {
using telemetry::dbl_gauge; // nop
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::doorman_ptr ptr) prometheus_broker::prometheus_broker(actor_config& cfg, io::doorman_ptr ptr)
...@@ -200,14 +51,6 @@ const char* prometheus_broker::name() const { ...@@ -200,14 +51,6 @@ const char* prometheus_broker::name() const {
return "caf.system.prometheus-broker"; 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() { behavior prometheus_broker::make_behavior() {
return { return {
[=](const io::new_data_msg& msg) { [=](const io::new_data_msg& msg) {
...@@ -267,17 +110,12 @@ behavior prometheus_broker::make_behavior() { ...@@ -267,17 +110,12 @@ behavior prometheus_broker::make_behavior() {
} }
void prometheus_broker::scrape() { void prometheus_broker::scrape() {
#ifdef HAS_PROCESS_METRICS // Scrape system metrics at most once per second.
// Collect system metrics at most once per second.
auto now = time(NULL); auto now = time(NULL);
if (last_scrape_ >= now) if (last_scrape_ < now) {
return; last_scrape_ = now;
last_scrape_ = now; proc_importer_.update();
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
} }
} // namespace caf::detail } // 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