diff --git a/.gitignore b/.gitignore index 0f35590110b27806e3f3d8e9c9d49251eb6d2f2f..14f4f8683221d834e87247d390e81df0bd4a1ea5 100644 --- a/.gitignore +++ b/.gitignore @@ -97,3 +97,4 @@ builds/ CMakeUserPresets.json gkfs/ .gitlab-ci-local/ +install-*-test/ diff --git a/README.md b/README.md index 730d2d36e0130be915a48c6b0cb3e227f5a7644f..c7fed24cd6edae86fe793f6a4d0df060459645fc 100644 --- a/README.md +++ b/README.md @@ -392,14 +392,30 @@ the `default_zmq` dependency profile. In addition, GekkoFS must be compiled with client metrics enabled (disabled by default) via the CMake argument `-DGKFS_ENABLE_CLIENT_METRICS=ON`. -Client metrics are individually enabled per GekkoFS client process via the following environment variables: +Client metrics/messages are enabled with common `GKFS_` environment variables. Exporting the same variables to both +clients and daemons enables client-side message emission and, when requested, the daemon-side relay: -- `LIBGKFS_ENABLE_METRICS=ON` enables capturing client-side metrics. +- `GKFS_ENABLE_METRICS=ON` enables capturing client-side metrics/messages. Daemons also honor this same variable when + deciding whether to start the optional relay, so no daemon-specific enable variable is needed. - `LIBGKFS_METRICS_FLUSH_INTERVAL=10` sets the flush interval to 10 seconds (defaults to 5). All outstanding client metrics are flushed when the process ends. - `LIBGKFS_METRICS_PATH=` sets the path to flush client-metrics (defaults to `/tmp/gkfs_client_metrics`). -- `LIBGKFS_METRICS_IP_PORT=127.0.0.1:5555` enables flushing to a set ZeroMQ server. This option disables flushing to a +- `GKFS_METRICS_IP_PORT=127.0.0.1:5555` enables flushing to a set ZeroMQ server. This option disables flushing to a file. +- `LIBGKFS_METRICS_IO_TYPE=w` restricts which of the two `ClientMetrics` instances (write, read) get created at + all. Defaults to `wr` (both, unchanged behavior). A write-only workload that never reads back what it wrote can + set this to `w` to skip the read-side thread and socket entirely, since a receiver that only consumes + `io_type=="w"` messages (e.g. FTIO's default prediction mode) would otherwise unpack and discard every read + message anyway. +- `GKFS_METRICS_AGGREGATOR=on` routes a client's metrics through the local per-node daemon instead of dialing the + ZeroMQ sink directly. At high rank density, every rank normally opens its own connection to one remote sink node; + this collapses that fan-in to one connection per node, and the daemon batches whatever it receives within a short + window into a single message before forwarding -- so the sink also receives O(nodes) messages instead of + O(ranks). The daemon needs the same `GKFS_ENABLE_METRICS`, `GKFS_METRICS_AGGREGATOR`, and + `GKFS_METRICS_IP_PORT` set in its own environment to actually start the aggregator thread. The receiving end needs + no configuration: a batched message + is a single top-level msgpack array, a direct message is a flat sequence of scalar fields, and the two are told + apart by inspecting the first unpacked object's type. The ZeroMQ export can be tested via the `gkfs_clientmetrics2json` application which is built when enabling the CMake option `-DGKFS_BUILD_TOOLS=ON`: @@ -407,7 +423,7 @@ option `-DGKFS_BUILD_TOOLS=ON`: - Starting the ZeroMQ server: `gkfs_clientmetrics2json tcp://127.0.0.1:5555` - `gkfs_clientmetrics2json ` can also be used to unpack the Messagepack export from a file. Examplarily output with the ZeroMQ sink enabled when running: - `LD_PRELOAD=libgkfs_intercept.so LIBGKFS_ENABLE_METRICS=ON LIBGKFS_METRICS_IP_PORT=127.0.0.1:5555 gkfs cp testfile /tmp/gkfs_mountdir/testfile`: + `LD_PRELOAD=libgkfs_intercept.so GKFS_ENABLE_METRICS=ON GKFS_METRICS_IP_PORT=127.0.0.1:5555 gkfs cp testfile /tmp/gkfs_mountdir/testfile`: ```bash ~ $ gkfs_clientmetrics2json tcp://127.0.0.1:5555 @@ -720,6 +736,11 @@ Client-metrics require the CMake argument `-DGKFS_ENABLE_CLIENT_METRICS=ON` (see - `LIBGKFS_METRICS_FLUSH_INTERVAL` - Set the flush interval for client metrics. - `LIBGKFS_METRICS_PATH` - Path to flush client metrics. - `LIBGKFS_METRICS_IP_PORT` - Enable flushing to a set ZeroMQ server (replaces `LIBGKFS_METRICS_PATH`). +- `LIBGKFS_METRICS_IO_TYPE` - Restrict metrics collection to `w`, `r`, or `wr` (default). Skips creating the + unused side entirely rather than collecting and discarding it. +- `LIBGKFS_METRICS_AGGREGATOR` - Route metrics through the local daemon (one connection per node) instead of + connecting to the ZeroMQ sink directly from every rank. Requires the same variable set in the daemon's + environment. - `LIBGKFS_PROXY_PID_FILE` - Path to the proxy pid file (when using the GekkoFS proxy). - `LIBGKFS_NUM_REPL` - Number of replicas for data. #### Optimization diff --git a/include/client/env.hpp b/include/client/env.hpp index e4fb7b9914e00fb2173465c9320f6bff4c0dc0f2..b5425f2c62644ff8c4de312e6227d4d0110a7db0 100644 --- a/include/client/env.hpp +++ b/include/client/env.hpp @@ -41,6 +41,7 @@ #define GKFS_CLIENT_ENV #include +#include #define ADD_PREFIX(str) CLIENT_ENV_PREFIX str @@ -61,11 +62,18 @@ static constexpr auto CWD = ADD_PREFIX("CWD"); static constexpr auto HOSTS_FILE = ADD_PREFIX("HOSTS_FILE"); static constexpr auto FORWARDING_MAP_FILE = ADD_PREFIX("FORWARDING_MAP_FILE"); #ifdef GKFS_ENABLE_CLIENT_METRICS -static constexpr auto ENABLE_METRICS = ADD_PREFIX("ENABLE_METRICS"); static constexpr auto METRICS_FLUSH_INTERVAL = ADD_PREFIX("METRICS_FLUSH_INTERVAL"); static constexpr auto METRICS_PATH = ADD_PREFIX("METRICS_PATH"); -static constexpr auto METRICS_IP_PORT = ADD_PREFIX("METRICS_IP_PORT"); +// "w", "r", or "wr"/"rw" (default) -- which ClientMetrics instances to +// create at all. A consumer that only ever reads io_type=="w" messages +// (e.g. FTIO's default prediction mode) shouldn't pay for a read-metrics +// thread+socket per rank that gets discarded downstream every time. +static constexpr auto METRICS_IO_TYPE = ADD_PREFIX("METRICS_IO_TYPE"); +// Opt-in, test-only: when set, connect to the local daemon's relay socket +// instead of dialing FTIO directly -- collapses per-rank connections down +// to one per node. The daemon must have GKFS_ENABLE_METRICS and the same +// routing vars set to actually start the relay (see daemon/env.hpp). #endif static constexpr auto PROTECT_FD = ADD_PREFIX("PROTECT_FD"); diff --git a/include/common/env.hpp b/include/common/env.hpp new file mode 100644 index 0000000000000000000000000000000000000000..c5ce6ad6f682852f64cf165034f4ecf805feecf0 --- /dev/null +++ b/include/common/env.hpp @@ -0,0 +1,59 @@ +/* + Copyright 2018-2025, Barcelona Supercomputing Center (BSC), Spain + Copyright 2015-2025, Johannes Gutenberg Universitaet Mainz, Germany + + This software was partially supported by the + EC H2020 funded project NEXTGenIO (Project ID: 671951, www.nextgenio.eu). + + This software was partially supported by the + ADA-FS project under the SPPEXA project funded by the DFG. + + This software was partially supported by the + the European Union’s Horizon 2020 JTI-EuroHPC research and + innovation programme, by the project ADMIRE (Project ID: 956748, + admire-eurohpc.eu) + + This project was partially promoted by the Ministry for Digital Transformation + and the Civil Service, within the framework of the Recovery, + Transformation and Resilience Plan - Funded by the European Union + -NextGenerationEU. + + This file is part of GekkoFS. + + GekkoFS is free software: you can redistribute it and/or modify + it under the terms of the GNU General Public License as published by + the Free Software Foundation, either version 3 of the License, or + (at your option) any later version. + + GekkoFS is distributed in the hope that it will be useful, + but WITHOUT ANY WARRANTY; without even the implied warranty of + MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + GNU General Public License for more details. + + You should have received a copy of the GNU General Public License + along with GekkoFS. If not, see . + + SPDX-License-Identifier: GPL-3.0-or-later +*/ + +#ifndef GKFS_COMMON_ENV_HPP +#define GKFS_COMMON_ENV_HPP + +#include + +#define ADD_PREFIX(str) COMMON_ENV_PREFIX str + +/* Environment variables shared by GekkoFS clients and daemons */ +namespace gkfs::env { + +#ifdef GKFS_ENABLE_CLIENT_METRICS +static constexpr auto ENABLE_METRICS = ADD_PREFIX("ENABLE_METRICS"); +static constexpr auto METRICS_IP_PORT = ADD_PREFIX("METRICS_IP_PORT"); +static constexpr auto METRICS_AGGREGATOR = ADD_PREFIX("METRICS_AGGREGATOR"); +#endif + +} // namespace gkfs::env + +#undef ADD_PREFIX + +#endif // GKFS_COMMON_ENV_HPP \ No newline at end of file diff --git a/include/common/msgpack_util.hpp b/include/common/msgpack_util.hpp index a4d86d044ea45e7dcc5333f884a933ae86da7265..fd2b43e1e023805ba3d1a7634791aeca50dffcd8 100644 --- a/include/common/msgpack_util.hpp +++ b/include/common/msgpack_util.hpp @@ -142,6 +142,12 @@ public: void zmq_connect(const std::string& ip_port); + // Like zmq_connect(), but takes a full zmq endpoint URI as-is (no + // "tcp://" prefix added) -- used for the LIBGKFS_METRICS_DAEMON_RELAY + // local ipc:// endpoint. + void + zmq_connect_raw(const std::string& endpoint); + bool zmq_is_connected(); diff --git a/include/config.hpp b/include/config.hpp index 47f7b6b8b836ad3d13568b49b38361327e279678..1f446c0215dab9cd3e1eedda3b2f2cb5b69855a8 100644 --- a/include/config.hpp +++ b/include/config.hpp @@ -115,7 +115,7 @@ constexpr auto dir = "metadata"; // Blocks are used to store the rename status (-1 is a renamed file) constexpr auto use_atime = false; constexpr auto use_ctime = true; -constexpr auto use_mtime = false; +constexpr auto use_mtime = true; constexpr auto use_link_cnt = false; constexpr auto use_blocks = true; /* diff --git a/include/daemon/env.hpp b/include/daemon/env.hpp index ce81c13fcdb5d3489513e6186300758583b77a04..e1bccf4112f6335ded902cd86008ce58d9ed2602 100644 --- a/include/daemon/env.hpp +++ b/include/daemon/env.hpp @@ -44,6 +44,7 @@ #define GKFS_DAEMON_ENV #include +#include #define ADD_PREFIX(str) COMMON_ENV_PREFIX str diff --git a/src/client/fuse/fuse_client.cpp b/src/client/fuse/fuse_client.cpp index fd1ddf3038cebf169bfc490ac959b68055070fb3..87ff84e864b4db8ce0a447554c270d28671a0974 100644 --- a/src/client/fuse/fuse_client.cpp +++ b/src/client/fuse/fuse_client.cpp @@ -473,8 +473,11 @@ read_handler(fuse_req_t req, fuse_ino_t ino, size_t size, off_t off, #ifdef GKFS_ENABLE_CLIENT_METRICS auto start_t = std::chrono::high_resolution_clock::now(); int rc = gkfs::syscall::gkfs_pread(fi->fh, buf.get(), size, off); - if(rc > 0) - CTX->read_metrics()->add_event(rc, start_t); + if(rc > 0) { + // may be null: LIBGKFS_METRICS_IO_TYPE can disable read-side metrics + if(auto rm = CTX->read_metrics()) + rm->add_event(rc, start_t); + } #else int rc = gkfs::syscall::gkfs_pread(fi->fh, buf.get(), size, off); #endif @@ -500,8 +503,11 @@ write_handler(fuse_req_t req, fuse_ino_t ino, const char* buf, size_t size, #ifdef GKFS_ENABLE_CLIENT_METRICS auto start_t = std::chrono::high_resolution_clock::now(); int rc = gkfs::syscall::gkfs_pwrite(fi->fh, buf, size, off); - if(rc > 0) - CTX->write_metrics()->add_event(rc, start_t); + if(rc > 0) { + // may be null: LIBGKFS_METRICS_IO_TYPE can disable write-side metrics + if(auto wm = CTX->write_metrics()) + wm->add_event(rc, start_t); + } #else int rc = gkfs::syscall::gkfs_pwrite(fi->fh, buf, size, off); #endif @@ -1478,10 +1484,14 @@ main(int argc, char* argv[]) { #ifdef GKFS_ENABLE_CLIENT_METRICS LOG(INFO, "Flushing final metrics..."); - CTX->write_metrics()->flush_msgpack(); - CTX->read_metrics()->flush_msgpack(); - LOG(INFO, "Metrics flushed. Total flush operations: {}", - CTX->write_metrics()->flush_count()); + // either may be null: LIBGKFS_METRICS_IO_TYPE can disable either side + if(auto wm = CTX->write_metrics()) { + wm->flush_msgpack(); + LOG(INFO, "Metrics flushed. Total flush operations: {}", + wm->flush_count()); + } + if(auto rm = CTX->read_metrics()) + rm->flush_msgpack(); CTX->destroy_metrics(); #endif diff --git a/src/client/gkfs_data.cpp b/src/client/gkfs_data.cpp index 7dc8292a736dccca5bde665eb6229145cc162a68..748c4df1fd452776af0e30a4263f1a08ec329925 100644 --- a/src/client/gkfs_data.cpp +++ b/src/client/gkfs_data.cpp @@ -345,7 +345,9 @@ gkfs_write_ws(gkfs::filemap::OpenFile& file, const char* buf, size_t count, #ifdef GKFS_ENABLE_CLIENT_METRICS auto start_t = std::chrono::high_resolution_clock::now(); auto written = gkfs_do_write(file, buf, count, offset, update_pos); - CTX->write_metrics()->add_event(written, start_t); + // may be null: LIBGKFS_METRICS_IO_TYPE can disable write-side metrics + if(auto wm = CTX->write_metrics()) + wm->add_event(written, start_t); #else auto written = gkfs_do_write(file, buf, count, offset, update_pos); #endif @@ -609,7 +611,9 @@ gkfs_read_ws(const gkfs::filemap::OpenFile& file, char* buf, size_t count, #ifdef GKFS_ENABLE_CLIENT_METRICS auto start_t = std::chrono::high_resolution_clock::now(); auto read = gkfs_do_read(file, buf, count, offset); - CTX->read_metrics()->add_event(read, start_t); + // may be null: LIBGKFS_METRICS_IO_TYPE can disable read-side metrics + if(auto rm = CTX->read_metrics()) + rm->add_event(read, start_t); return read; #else return gkfs_do_read(file, buf, count, offset); diff --git a/src/client/preload.cpp b/src/client/preload.cpp index e51d860217bc8f77f4a320243cecbabaa0628721..fe3996a635b9043ed372d86b09c1f64db5ef6aa6 100644 --- a/src/client/preload.cpp +++ b/src/client/preload.cpp @@ -583,10 +583,14 @@ destroy_preload() { } #ifdef GKFS_ENABLE_CLIENT_METRICS LOG(INFO, "Flushing final metrics..."); - CTX->write_metrics()->flush_msgpack(); - CTX->read_metrics()->flush_msgpack(); - LOG(INFO, "Metrics flushed. Total flush operations: {}", - CTX->write_metrics()->flush_count()); + // either may be null: LIBGKFS_METRICS_IO_TYPE can disable either side + if(auto wm = CTX->write_metrics()) { + wm->flush_msgpack(); + LOG(INFO, "Metrics flushed. Total flush operations: {}", + wm->flush_count()); + } + if(auto rm = CTX->read_metrics()) + rm->flush_msgpack(); CTX->destroy_metrics(); #endif CTX->clear_hosts(); diff --git a/src/client/preload_context.cpp b/src/client/preload_context.cpp index 1cbf42093bbf6c553ecbde3379163f7643247433..89fef777ebf19661d41478b8d6b072077e8d0844 100644 --- a/src/client/preload_context.cpp +++ b/src/client/preload_context.cpp @@ -165,50 +165,90 @@ PreloadContext::init_metrics() { auto flush_interval = std::stoi(gkfs::env::get_var( gkfs::env::METRICS_FLUSH_INTERVAL, std::to_string(gkfs::config::client_metrics::flush_interval))); + // FTIO's default prediction mode only consumes io_type=="w" + // messages (see parse_gekko.py) -- every read-metrics message it + // receives is unpacked and then discarded. Left unfiltered, every rank + // still pays for a full second ClientMetrics instance (its own thread, + // socket, flush cycle) purely to be thrown away downstream. Default + // "wr" preserves prior behavior exactly; set LIBGKFS_METRICS_IO_TYPE=w + // to skip creating read_metrics_ entirely for write-only workloads. + auto io_type_filter = gkfs::env::get_var(gkfs::env::METRICS_IO_TYPE, "wr"); + bool want_write = io_type_filter.find('w') != std::string::npos; + bool want_read = io_type_filter.find('r') != std::string::npos; if(gkfs::env::var_is_set(gkfs::env::METRICS_IP_PORT)) { - write_metrics_ = std::make_unique( - gkfs::messagepack::client_metric_io_type::write, - gkfs::messagepack::client_metric_flush_type::socket, - flush_interval); - read_metrics_ = std::make_unique( - gkfs::messagepack::client_metric_io_type::read, - gkfs::messagepack::client_metric_flush_type::socket, - flush_interval); + if(want_write) + write_metrics_ = std::make_unique( + gkfs::messagepack::client_metric_io_type::write, + gkfs::messagepack::client_metric_flush_type::socket, + flush_interval); + if(want_read) + read_metrics_ = std::make_unique( + gkfs::messagepack::client_metric_io_type::read, + gkfs::messagepack::client_metric_flush_type::socket, + flush_interval); if(gkfs::env::var_is_set(gkfs::env::ENABLE_METRICS)) { LOG(INFO, "Client metrics enabled with ZeroMQ flushing. Initializing..."); - write_metrics_->enable(); - read_metrics_->enable(); auto metrics_ip = gkfs::env::get_var(gkfs::env::METRICS_IP_PORT); - write_metrics_->zmq_connect(metrics_ip); - if(!write_metrics_->zmq_is_connected()) { - LOG(ERROR, "Client write metrics failed to connect to : {}", + // test-only opt-in (LIBGKFS_METRICS_AGGREGATOR) -- + // connect to the local daemon's relay socket instead of FTIO + // directly, so many ranks converge on one node-local connection + // instead of each dialing FTIO's remote node individually. The + // daemon (daemon.cpp: start_metrics_relay()) must have the same + // enable/routing vars set to actually bind that socket. + bool via_relay = + gkfs::env::var_is_set(gkfs::env::METRICS_AGGREGATOR); + // error level is deliberate diagnostic-only visibility + // for this test-only feature -- the JIT harness hardcodes + // LIBGKFS_LOG=errors, so info-level never reaches the log file. + if(via_relay) + LOG(ERROR, + "Client metrics: via_relay=true, connecting to " + "local daemon relay instead of {}", metrics_ip); - return false; + if(want_write) { + write_metrics_->enable(); + if(via_relay) + write_metrics_->zmq_connect_raw( + "ipc:///tmp/gkfs_metrics_aggregator.sock"); + else + write_metrics_->zmq_connect(metrics_ip); + if(!write_metrics_->zmq_is_connected()) { + LOG(ERROR, "Client write metrics failed to connect to : {}", + metrics_ip); + return false; + } + LOG(INFO, "Client write metrics connected to : {}", metrics_ip); } - LOG(INFO, "Client write metrics connected to : {}", metrics_ip); - read_metrics_->zmq_connect(metrics_ip); - if(!read_metrics_->zmq_is_connected()) { - LOG(ERROR, "Client read metrics failed to connect to : {}", - metrics_ip); - return false; + if(want_read) { + read_metrics_->enable(); + if(via_relay) + read_metrics_->zmq_connect_raw( + "ipc:///tmp/gkfs_metrics_aggregator.sock"); + else + read_metrics_->zmq_connect(metrics_ip); + if(!read_metrics_->zmq_is_connected()) { + LOG(ERROR, "Client read metrics failed to connect to : {}", + metrics_ip); + return false; + } + LOG(INFO, "Client read metrics connected to : {}", metrics_ip); } - LOG(INFO, "Client read metrics connected to : {}", metrics_ip); } } else { - write_metrics_ = std::make_unique( - gkfs::messagepack::client_metric_io_type::write, - gkfs::messagepack::client_metric_flush_type::file, - flush_interval); - read_metrics_ = std::make_unique( - gkfs::messagepack::client_metric_io_type::read, - gkfs::messagepack::client_metric_flush_type::file, - flush_interval); + if(want_write) + write_metrics_ = std::make_unique( + gkfs::messagepack::client_metric_io_type::write, + gkfs::messagepack::client_metric_flush_type::file, + flush_interval); + if(want_read) + read_metrics_ = std::make_unique( + gkfs::messagepack::client_metric_io_type::read, + gkfs::messagepack::client_metric_flush_type::file, + flush_interval); if(gkfs::env::var_is_set(gkfs::env::ENABLE_METRICS)) { LOG(INFO, "Client metrics enabled with file flushing. Initializing..."); - write_metrics_->enable(); - read_metrics_->enable(); if(!gkfs::env::var_is_set(gkfs::env::METRICS_PATH)) { LOG(WARNING, "No metrics path set. Using default path at {}", gkfs::config::client_metrics::flush_path); @@ -217,10 +257,18 @@ PreloadContext::init_metrics() { gkfs::env::METRICS_PATH, gkfs::config::client_metrics::flush_path); std::filesystem::create_directories(metrics_path); - write_metrics_->path(metrics_path, "write"); - LOG(INFO, "Client write metrics path: {}", write_metrics_->path()); - read_metrics_->path(metrics_path, "read"); - LOG(INFO, "Client read metrics path: {}", read_metrics_->path()); + if(want_write) { + write_metrics_->enable(); + write_metrics_->path(metrics_path, "write"); + LOG(INFO, "Client write metrics path: {}", + write_metrics_->path()); + } + if(want_read) { + read_metrics_->enable(); + read_metrics_->path(metrics_path, "read"); + LOG(INFO, "Client read metrics path: {}", + read_metrics_->path()); + } } } #endif diff --git a/src/common/msgpack_util.cpp b/src/common/msgpack_util.cpp index 88f1bd6af62f6f27e0f4838420794aed4fd5bf87..9ea844cb8cf050651bec3e74e3a6f7ae5dc08b7f 100644 --- a/src/common/msgpack_util.cpp +++ b/src/common/msgpack_util.cpp @@ -83,6 +83,12 @@ ClientMetrics::ClientMetrics(client_metric_io_type io_type, zmq_flush_context_ = std::make_unique(1); zmq_flush_socket_ = std::make_unique(*zmq_flush_context_, ZMQ_PUSH); + // metrics are instrumentation, not correctness-critical -- + // never let a congested FTIO node stall the app. Default LINGER (-1) + // would make close() block indefinitely trying to deliver a queued + // message; 0 makes it drop and return immediately instead. + int linger = 0; + zmq_flush_socket_->setsockopt(ZMQ_LINGER, &linger, sizeof(linger)); } flush_thread_running_ = true; flush_thread_ = std::thread(&ClientMetrics::flush_loop, this); @@ -161,8 +167,11 @@ ClientMetrics::flush_msgpack() { zmq::message_t message(data.size()); // copy data from serialized msgpack to zmq message memcpy(message.data(), data.data(), data.size()); - // non-blocking zmq send - if(zmq_flush_socket_->send(message, zmq::send_flags::none) == -1) { + // this comment previously said "non-blocking" but passed + // send_flags::none, which is the blocking default -- with SNDHWM=1000 + // and hundreds of ranks converging on one FTIO node, that can genuinely + // stall the caller. dontwait makes it actually non-blocking. + if(zmq_flush_socket_->send(message, zmq::send_flags::dontwait) == -1) { std::cerr << "Failed to send zmq message\n"; } } @@ -198,6 +207,11 @@ ClientMetrics::zmq_connect(const string& ip_port) { zmq_flush_socket_->connect(address); } +void +ClientMetrics::zmq_connect_raw(const string& endpoint) { + zmq_flush_socket_->connect(endpoint); +} + bool ClientMetrics::zmq_is_connected() { return zmq_flush_socket_->handle() != nullptr; diff --git a/src/daemon/CMakeLists.txt b/src/daemon/CMakeLists.txt index 65af3945cae13722c9113fce74b474678c25107f..1ba881612482b031dac3b62e86c887860d55dcf8 100644 --- a/src/daemon/CMakeLists.txt +++ b/src/daemon/CMakeLists.txt @@ -74,6 +74,12 @@ target_link_libraries( ZStd::ZStd ) +if (GKFS_ENABLE_CLIENT_METRICS) + # for the opt-in LIBGKFS_METRICS_AGGREGATOR test feature (daemon.cpp) + target_link_libraries(gkfs_daemon PRIVATE cppzmq Msgpack::Msgpack) + target_compile_definitions(gkfs_daemon PUBLIC GKFS_ENABLE_CLIENT_METRICS) +endif () + if (GKFS_ENABLE_AGIOS) target_sources(gkfs_daemon PRIVATE scheduler/agios.cpp) target_compile_definitions(gkfs_daemon PUBLIC GKFS_ENABLE_AGIOS) diff --git a/src/daemon/daemon.cpp b/src/daemon/daemon.cpp index 5470dab20fb041e89091892f685694ba3000f6e0..a2fd49a35b90ce509d2e92e358bdceb881ebc3f6 100644 --- a/src/daemon/daemon.cpp +++ b/src/daemon/daemon.cpp @@ -72,6 +72,8 @@ #include #include #include +#include +#include extern "C" { #include @@ -80,6 +82,11 @@ extern "C" { #include +#ifdef GKFS_ENABLE_CLIENT_METRICS +#include +#include +#endif + using namespace std; namespace fs = std::filesystem; namespace tl = thallium; @@ -88,6 +95,107 @@ static condition_variable shutdown_please; // handler for shutdown signaling static mutex mtx; // mutex to wait on shutdown conditional variable static bool keep_rootdir = true; +#ifdef GKFS_ENABLE_CLIENT_METRICS +namespace { +// test-only opt-in feature (GKFS_ENABLE_METRICS + GKFS_METRICS_AGGREGATOR). +// Every +// rank normally dials FTIO's node directly for its own metrics flush -- +// at high rank density that's hundreds of connections converging on one +// node, each sending its own small message. This collapses both: local +// ranks push to this daemon over a unix socket instead of the network, +// and the daemon batches whatever arrives within one BATCH_WINDOW_MS +// window into a single msgpack array-of-bytes message before forwarding +// to FTIO -- so the network leg also drops from O(ranks) messages to +// O(nodes * windows) messages, not just O(nodes) connections. FTIO's +// receiver (ftio/prediction/processes_zmq.py: unbatch_messages()) needs +// no flag -- it auto-detects a batch (one top-level msgpack array) vs a +// direct message (flat 8-9 top-level scalars) by peeking at the first +// unpacked object's type. +static const char* METRICS_AGGREGATOR_ENDPOINT = + "ipc:///tmp/gkfs_metrics_aggregator.sock"; +static constexpr int BATCH_WINDOW_MS = 500; +static std::atomic metrics_relay_running{false}; +static std::thread metrics_relay_thread; + +void +metrics_relay_loop(std::string upstream_addr) { + zmq::context_t ctx(1); + zmq::socket_t pull(ctx, ZMQ_PULL); + zmq::socket_t push(ctx, ZMQ_PUSH); + int timeout_ms = BATCH_WINDOW_MS; + int linger = 0; + pull.setsockopt(ZMQ_RCVTIMEO, &timeout_ms, sizeof(timeout_ms)); + pull.setsockopt(ZMQ_LINGER, &linger, sizeof(linger)); + push.setsockopt(ZMQ_LINGER, &linger, sizeof(linger)); + pull.bind(METRICS_AGGREGATOR_ENDPOINT); + push.connect("tcp://" + upstream_addr); + + std::vector batch; + size_t batches_sent = 0; + size_t records_sent = 0; + auto flush_batch = [&]() { + if(batch.empty()) + return; + msgpack::Packer packer; + packer(batch); + const auto& bytes = packer.vector(); + zmq::message_t out(bytes.size()); + memcpy(out.data(), bytes.data(), bytes.size()); + push.send(out, zmq::send_flags::dontwait); + records_sent += batch.size(); + // diagnostic-only (see start_metrics_relay comment) -- + // log the first batch and then every 20th, so a quick test run + // shows whether anything is actually flowing without spamming. + if(++batches_sent == 1 || batches_sent % 20 == 0) + GKFS_DATA->spdlogger()->error( + "Metrics aggregator: sent batch #{} ({} records, " + "{} total)", + batches_sent, batch.size(), records_sent); + batch.clear(); + }; + + while(metrics_relay_running.load()) { + zmq::message_t msg; + auto res = pull.recv(msg, zmq::recv_flags::none); + if(!res.has_value()) { + flush_batch(); // RCVTIMEO elapsed: window closed, send what we have + continue; + } + batch.emplace_back(static_cast(msg.data()), msg.size()); + } + flush_batch(); // drain on shutdown so the last window isn't lost +} + +void +start_metrics_relay() { + if(!gkfs::env::var_is_set(gkfs::env::ENABLE_METRICS) || + !gkfs::env::var_is_set(gkfs::env::METRICS_AGGREGATOR) || + !gkfs::env::var_is_set(gkfs::env::METRICS_IP_PORT)) + return; + auto upstream = gkfs::env::get_var(gkfs::env::METRICS_IP_PORT); + ::unlink("/tmp/gkfs_metrics_aggregator.sock"); // stale socket from a crash + metrics_relay_running.store(true); + metrics_relay_thread = std::thread(metrics_relay_loop, upstream); + // error level (not info) is deliberate -- the JIT harness + // hardcodes GKFS_DAEMON_LOG_LEVEL=err, so info never reaches the log + // file. This is diagnostic-only visibility for a test-only feature, + // not a real error. + GKFS_DATA->spdlogger()->error( + "Metrics daemon-relay enabled, forwarding {} -> {}", + METRICS_AGGREGATOR_ENDPOINT, upstream); +} + +void +stop_metrics_relay() { + if(!metrics_relay_running.load()) + return; + metrics_relay_running.store(false); + if(metrics_relay_thread.joinable()) + metrics_relay_thread.join(); +} +} // namespace +#endif + namespace { struct cli_options { string mountdir; @@ -555,6 +663,10 @@ init_environment() { } GKFS_DATA->spdlogger()->debug("{}() MalleableManager running.", __func__); +#ifdef GKFS_ENABLE_CLIENT_METRICS + start_metrics_relay(); +#endif + GKFS_DATA->spdlogger()->info("Startup successful. Daemon is ready."); } @@ -589,6 +701,9 @@ agios_initialize() { void destroy_enviroment() { std::error_code ecode; +#ifdef GKFS_ENABLE_CLIENT_METRICS + stop_metrics_relay(); +#endif GKFS_DATA->spdlogger()->debug("{}() Freeing I/O executions streams", __func__); for(unsigned int i = 0; i < RPC_DATA->io_streams().size(); i++) {