diff --git a/docs/telemetry.md b/docs/telemetry.md index 077a4f03b0..1020c53191 100644 --- a/docs/telemetry.md +++ b/docs/telemetry.md @@ -14,7 +14,8 @@ Custom telemetry exporter plug-ins can be created according to [src/plugins/tele 2. **Shared Memory Buffer**: Statically-linked built in implementation of telemetry exporter. Uses shared memory cyclic buffer for efficient event storage and export. 3. **Telemetry Readers**: C++ and Python applications to read and display telemetry data from the cyclic buffer. 4. **Prometheus exporter**: EXPERIMENTAL (beta) Prometheus compatible telemetry exporter, see [src/plugins/telemetry/prometheus/README.md](../src/plugins/telemetry/prometheus/README.md). -5. **DOCA exporter**: EXPERIMENTAL DOCA/CollectX telemetry exporter. Drives one or more delivery backends (`NIXL_TELEMETRY_DOCA_BACKENDS`, default `scrape`): a local Prometheus scrape endpoint and/or `ipc` push to the DOCA Telemetry Service (DTS) for single-endpoint multi-process aggregation. See [src/plugins/telemetry/doca/README.md](../src/plugins/telemetry/doca/README.md). +5. **Multi-process Prometheus exporter (`prometheus_mp`)**: EXPERIMENTAL native aggregation of all processes of a multi-process run behind one scrape endpoint, without an external service. See [src/plugins/telemetry/prometheus_mp/README.md](../src/plugins/telemetry/prometheus_mp/README.md) and "Multi-process aggregation" below. +6. **DOCA exporter**: EXPERIMENTAL DOCA/CollectX telemetry exporter. Drives one or more delivery backends (`NIXL_TELEMETRY_DOCA_BACKENDS`, default `scrape`): a local Prometheus scrape endpoint and/or `ipc` push to the DOCA Telemetry Service (DTS) for single-endpoint multi-process aggregation. See [src/plugins/telemetry/doca/README.md](../src/plugins/telemetry/doca/README.md). ### Event Structure @@ -147,7 +148,55 @@ Telemetry is configured by environment variables: - When telemetry is requested but no output sink is configured (neither `NIXL_TELEMETRY_EXPORTER` nor `NIXL_TELEMETRY_DIR`), it falls back to the collect-only NOP exporter: events are collected in-process so `getXferTelemetry()` / `get_xfer_telemetry()` works, but nothing is written out. - If telemetry is enabled but no exporter is set, or the exporter name is empty, then the sink depends on `NIXL_TELEMETRY_DIR` as explained below (falling back to NOP when it is unset). - Set `NIXL_TELEMETRY_EXPORTER=NOP` to explicitly keep telemetry active (events are collected and `getXferTelemetry()` works) while discarding all output. It needs no sink and writes nothing, so it can be used to measure the overhead of the telemetry collection path in isolation. -- Exporters that expose a scrape endpoint (e.g. Prometheus) bind one port per process. Under multi-process runs (e.g. tensor/data parallelism) every rank tries to bind the same port; only one wins. Losing that race is benign and non-fatal: the affected process logs a single warning and runs without a telemetry sink instead of failing agent construction. See [src/plugins/telemetry/prometheus/README.md](../src/plugins/telemetry/prometheus/README.md). +- Exporters that expose a scrape endpoint (e.g. Prometheus) bind one port per process. Under multi-process runs (e.g. tensor/data parallelism) every rank tries to bind the same port; only one wins. With the single-process `prometheus` exporter, losing that race is benign and non-fatal: the affected process logs a single warning and runs without a telemetry sink instead of failing agent construction (see [src/plugins/telemetry/prometheus/README.md](../src/plugins/telemetry/prometheus/README.md)). To instead export **every** rank behind one endpoint, use the `prometheus_mp` exporter (see "Multi-process aggregation" below). + +## Multi-process aggregation (`prometheus_mp`) + +The `prometheus_mp` exporter aggregates the telemetry of all processes of a +multi-process NIXL run behind a **single** Prometheus scrape endpoint, natively +(no DOCA/DTS). Every process writes its own metric state to a per-process +memory-mapped file in a shared directory; the processes race to bind the scrape +port, the winner serves `/metrics` by reading and republishing all live processes' +files on each scrape (labeled per process), and the losers run as writers only. A +bind collision is benign, so no rank is dropped. Full details: +[src/plugins/telemetry/prometheus_mp/README.md](../src/plugins/telemetry/prometheus_mp/README.md). + +### Configuration + +| Variable | Description | Default | +| -------- | ----------- | ------- | +| `NIXL_TELEMETRY_EXPORTER` | Set to `prometheus_mp` to select this exporter | - | +| `NIXL_TELEMETRY_MULTIPROC_DIR` | Shared directory for per-process store files (all ranks to be aggregated must use the same path). **Required.** | - | +| `NIXL_TELEMETRY_PROMETHEUS_PORT` | Scrape port (shared with the `prometheus` exporter) | `9090` | +| `NIXL_TELEMETRY_PROMETHEUS_LOCAL` | Bind `127.0.0.1` instead of `0.0.0.0` | `false` | +| `NIXL_TELEMETRY_RANK_ENV` | Name of the env var holding the rank for the optional `local_rank` label; no label if that env var is unset | `LOCAL_RANK` | +| `NIXL_TELEMETRY_MP_STALE_TTL` | Seconds after a dead process's last update before its store is stale and reaped | `30` | + +Series are labeled by `hostname`, `agent_name`, `pid` (guarantees cross-process +uniqueness), `agent_instance` (distinguishes multiple same-name agents within one +process), and optionally `local_rank` (the local/per-GPU rank, not Dynamo's +data-parallel `dp_rank`). `agent_errors_total` additionally includes the bounded +`status` label. Metric names, types, and semantics are identical to the +single-process `prometheus` exporter. + +`NIXL_TELEMETRY_MULTIPROC_DIR` follows Dynamo's `PROMETHEUS_MULTIPROC_DIR` +convention: a shared **local** folder (not NFS), one per pod / process-family, +treated as ephemeral (e.g. a per-pod Kubernetes `emptyDir`). Because NIXL is a +library loaded independently per rank -- with no parent to propagate the path as in +Dynamo -- the launcher/operator must set the **same** directory for every rank, so +it is required rather than auto-defaulted. + +### Scope & limitations + +`prometheus_mp` is purpose-built for NIXL's telemetry model, **not** a generic +Prometheus multiprocess store, and it does not reuse Python `prometheus_client`'s +multiprocess format. The metric set is fixed at compile time (positional slots; +names are not stored) and per-process label values are captured once at startup and +never change -- events carry only a numeric value, with no per-observation labels. +It therefore **cannot represent a metric with a dynamic / high-cardinality label** +that varies per observation. No NIXL metric needs that today; if one is ever added, +a different (keyed) store would be required. For aggregation via an external +service, use the DOCA/CollectX exporter instead. ## Cyclic Buffer diff --git a/src/plugins/telemetry/meson.build b/src/plugins/telemetry/meson.build index 4b05ebc04a..d17586a16f 100644 --- a/src/plugins/telemetry/meson.build +++ b/src/plugins/telemetry/meson.build @@ -14,6 +14,7 @@ # limitations under the License. subdir('prometheus') +subdir('prometheus_mp') # DOCA telemetry exporter: an optional DOCA pkg-config dependency, gated exactly # like the GPUNETIO backend (src/plugins/meson.build). Built by default when the diff --git a/src/plugins/telemetry/prometheus_mp/README.md b/src/plugins/telemetry/prometheus_mp/README.md new file mode 100644 index 0000000000..e0142b4649 --- /dev/null +++ b/src/plugins/telemetry/prometheus_mp/README.md @@ -0,0 +1,134 @@ + + +# NIXL multi-process Prometheus Telemetry exporter plug-in (`prometheus_mp`) + +`prometheus_mp` exposes the telemetry of **all** processes of a multi-process +NIXL run (e.g. tensor/data parallelism) behind a **single** Prometheus scrape +endpoint, without any DOCA/DTS dependency. + +It complements the single-process [`prometheus`](../prometheus/README.md) exporter +(which binds one port per process, so only one rank's metrics are scraped) and the +DOCA/CollectX exporter (which aggregates via an external DTS service). Use +`prometheus_mp` when you want all ranks aggregated natively with no extra +infrastructure. General NIXL telemetry background: [docs/telemetry.md](../../../../docs/telemetry.md). + +## Dependencies + +Same as the `prometheus` plug-in: the bundled prometheus-cpp subproject and +`libcurl` (`libcurl4-openssl-dev` / `libcurl-devel`). + +## How it works + +- **Every process writes its own metric state** to a per-process memory-mapped + file in a shared directory (`NIXL_TELEMETRY_MULTIPROC_DIR`). Updates are + lock-free; there is no serialization. +- **Bind-race owner election.** On startup each process tries to bind the scrape + port. The one that wins ("owner") runs the HTTP endpoint plus a collector that, + on each scrape, reads every live process's file and republishes them as one + exposition. The processes that lose the race run in **writer-only** mode (no + HTTP server). A bind collision is therefore benign -- every process gets a valid + telemetry sink; no rank is dropped and no scary error is logged. +- **Per-process series.** Each process is exported as its own series (cumulative + counters and last-operation gauges), never summed across processes, so + per-process values stay correct and monotonic. +- **Stale handling.** On clean shutdown a process removes its own store file. If + it instead crashes or is killed -- and so cannot clean up after itself -- the + owner drops its series once the process is gone (verified by pid + `/proc` start + time) or its data ages past the TTL, and reaps the file. + +## Configuration + +```bash +export NIXL_TELEMETRY_ENABLE="y" +export NIXL_TELEMETRY_EXPORTER="prometheus_mp" # selects libtelemetry_exporter_prometheus_mp.so +export NIXL_TELEMETRY_MULTIPROC_DIR="/tmp/nixl_metrics" # REQUIRED: shared by all ranks in the pod +``` + +This mirrors Dynamo's `PROMETHEUS_MULTIPROC_DIR` convention (a shared folder that +every related process writes into, one leader exports): all ranks that should be +aggregated together must point `NIXL_TELEMETRY_MULTIPROC_DIR` at the **same** +directory. Unlike Dynamo -- which auto-creates a temp dir in the parent and lets +child engine processes inherit it -- NIXL is a library loaded independently in each +rank, so there is no parent to propagate the path; the launcher/operator must set +the same directory for every rank (hence it is required, not auto-defaulted, so a +per-process temp dir can never silently break aggregation). + +Recommended, following Dynamo's model: a shared **local** folder, one per pod / +process-family, treated as ephemeral (e.g. a per-pod Kubernetes `emptyDir`, or a +temp dir cleaned between runs). It must be a local filesystem -- **not** a network +filesystem (NFS/CIFS), where mmap `MAP_SHARED` cross-process visibility is not +guaranteed (the same restriction Dynamo's multiprocess dir has). tmpfs (e.g. a +Memory-medium `emptyDir` or `/dev/shm`) works and avoids any disk writeback, but is +optional -- a plain local dir is fine, since updates hit the page cache and the +per-process store files are ~one page each. + +### Optional configuration + +```bash +# Scrape port (default 9090) and bind scope -- shared with the prometheus plug-in. +export NIXL_TELEMETRY_PROMETHEUS_PORT="" +export NIXL_TELEMETRY_PROMETHEUS_LOCAL="y" # bind 127.0.0.1 instead of 0.0.0.0 + +# Optional local_rank label: names the env var that holds the rank (default LOCAL_RANK). +# If that env var is unset, no local_rank label is emitted (series stay unique via pid). +export NIXL_TELEMETRY_RANK_ENV="LOCAL_RANK" + +# Seconds after a dead process's last update before its store is considered stale +# and reaped (default 30). A live process is always published regardless of age. +export NIXL_TELEMETRY_MP_STALE_TTL="30" +``` + +## Metric labels + +Every series is labeled by: + +- `hostname` -- host where the agent runs. +- `agent_name` -- the agent name given at initialization. +- `pid` -- the producing process id. This guarantees each process is a distinct + series even if agent names collide; it is deliberately **not** named `instance` + (a reserved Prometheus target label). +- `agent_instance` -- a per-process counter distinguishing multiple agents created + in the same process (which share `pid`, `hostname`, and `agent_name`), so their + series never collide. `0` for the common single-agent-per-process case. +- `local_rank` -- **optional**, present only when a rank env var (see + `NIXL_TELEMETRY_RANK_ENV`) is set. This is the local/per-GPU (TP) rank, distinct + from Dynamo's data-parallel `dp_rank`. +- `status` -- only on `agent_errors_total`, bounded by the fixed `AGENT_ERR_*` set. + +The metric names, types, counter/gauge semantics, and events are identical to the +single-process [`prometheus`](../prometheus/README.md) exporter (same shared +descriptor). + +## Design scope & limitations + +This exporter is **purpose-built for NIXL's telemetry model, not a generic +Prometheus multiprocess store** (in particular it is not compatible with, and does +not reuse, Python `prometheus_client`'s multiprocess format): + +- The metric set is fixed at compile time; slots are positional, so metric names + are never stored in the files. +- Per-process label values (`hostname`, `agent_name`, `pid`, `local_rank`) are + captured once at startup and never change. Events carry only a numeric value -- + there are **no per-observation labels**. +- Consequently the store **cannot represent a metric with a dynamic / + high-cardinality label** whose value varies per observation. No NIXL metric has + such a label today; if one is ever added, this exporter would need a different + (keyed) store. + +This is the native, dependency-free path. For aggregation through an external +telemetry service, use the DOCA/CollectX exporter (IPC to DTS) instead. diff --git a/src/plugins/telemetry/prometheus_mp/meson.build b/src/plugins/telemetry/prometheus_mp/meson.build new file mode 100644 index 0000000000..4d08ccda01 --- /dev/null +++ b/src/plugins/telemetry/prometheus_mp/meson.build @@ -0,0 +1,87 @@ +# SPDX-FileCopyrightText: Copyright (c) 2025-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +# Multiprocess metric-state store: a small, prometheus-cpp-free component (mmap +# per-process store) shared by the prometheus_mp plugin and its unit tests. Built +# unconditionally so the CTest coverage runs even where prometheus-cpp is absent; +# the plugin shared library (which needs prometheus-cpp) is added in a later step. +prometheus_mp_inc_dirs = include_directories('.') + +telemetry_mp_store_lib = static_library( + 'nixl_telemetry_mp_store', + 'mp_store.cpp', + include_directories: [nixl_inc_dirs, utils_inc_dirs, prometheus_mp_inc_dirs], + dependencies: [nixl_common_dep], + install: false, + pic: true, +) + +nixl_telemetry_mp_store_dep = declare_dependency( + link_with: telemetry_mp_store_lib, + include_directories: prometheus_mp_inc_dirs, + dependencies: [nixl_common_dep], +) + +# Scrape-time collector: converts peer store snapshots into prometheus-cpp metric +# families. Needs prometheus-cpp (core), so it is gated on it being available +# (same condition as the prometheus plugin, set in the sibling subdir above). +if prometheus_found + telemetry_mp_collector_lib = static_library( + 'nixl_telemetry_mp_collector', + 'mp_collector.cpp', + include_directories: [nixl_inc_dirs, utils_inc_dirs, prometheus_mp_inc_dirs], + dependencies: [nixl_common_dep, prometheus_core_dep, nixl_telemetry_mp_store_dep], + install: false, + pic: true, + ) + + nixl_telemetry_mp_collector_dep = declare_dependency( + link_with: telemetry_mp_collector_lib, + include_directories: prometheus_mp_inc_dirs, + dependencies: [nixl_common_dep, prometheus_core_dep, nixl_telemetry_mp_store_dep], + ) + + # Exporter with bind-race election. Built as a static library too so it can be + # unit-tested directly (in addition to being packaged into the plugin .so). + telemetry_mp_exporter_lib = static_library( + 'nixl_telemetry_mp_exporter', + 'prometheus_mp_exporter.cpp', + include_directories: [nixl_inc_dirs, utils_inc_dirs, prometheus_mp_inc_dirs], + dependencies: [nixl_common_dep, prometheus_core_dep, prometheus_pull_dep, + nixl_telemetry_mp_store_dep, nixl_telemetry_mp_collector_dep], + install: false, + pic: true, + ) + + nixl_telemetry_mp_exporter_dep = declare_dependency( + link_with: telemetry_mp_exporter_lib, + include_directories: prometheus_mp_inc_dirs, + dependencies: [nixl_common_dep, prometheus_core_dep, prometheus_pull_dep, + nixl_telemetry_mp_store_dep, nixl_telemetry_mp_collector_dep], + ) + + # Installable plugin: selected via NIXL_TELEMETRY_EXPORTER=prometheus_mp. + prometheus_mp_exporter_plugin = shared_library( + 'libtelemetry_exporter_prometheus_mp', + 'prometheus_mp_plugin.cpp', + include_directories: [nixl_inc_dirs, utils_inc_dirs, prometheus_mp_inc_dirs], + dependencies: [nixl_infra, nixl_common_dep, absl_log_dep, prometheus_core_dep, + prometheus_pull_dep, nixl_telemetry_mp_exporter_dep], + install: true, + install_dir: plugin_install_dir, + name_prefix: '', + ) + message('Building Prometheus multiprocess telemetry exporter plugin') +endif diff --git a/src/plugins/telemetry/prometheus_mp/mp_collector.cpp b/src/plugins/telemetry/prometheus_mp/mp_collector.cpp new file mode 100644 index 0000000000..d7c73aab3d --- /dev/null +++ b/src/plugins/telemetry/prometheus_mp/mp_collector.cpp @@ -0,0 +1,246 @@ +/* + * SPDX-FileCopyrightText: Copyright (c) 2025-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. + * SPDX-License-Identifier: Apache-2.0 + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +#include "mp_collector.h" + +#include "common/nixl_log.h" +#include "common/nixl_time.h" +#include "telemetry_event.h" + +#include +#include + +#include +#include +#include + +#include +#include +#include +#include +#include + +namespace nixl::telemetry::mp { + +namespace { + + using prometheus::ClientMetric; + using prometheus::MetricFamily; + using prometheus::MetricType; + + [[nodiscard]] std::vector + baseLabels(const mpStoreSnapshot &s) { + std::vector labels; + labels.push_back({"hostname", s.hostname}); + labels.push_back({"agent_name", s.agentName}); + // pid guarantees per-process series uniqueness even if agent names are + // not unique across processes; avoids duplicate-series scrape errors. Not + // named "instance" (a reserved Prometheus target label). + labels.push_back({"pid", std::to_string(s.pid)}); + labels.push_back({"agent_instance", std::to_string(s.instance)}); + if (!s.localRank.empty()) { + labels.push_back({"local_rank", s.localRank}); + } + return labels; + } + + [[nodiscard]] ClientMetric + counterMetric(std::vector labels, uint64_t value) { + ClientMetric m; + m.label = std::move(labels); + m.counter.value = static_cast(value); + return m; + } + + [[nodiscard]] ClientMetric + gaugeMetric(std::vector labels, uint64_t value) { + ClientMetric m; + m.label = std::move(labels); + m.gauge.value = static_cast(value); + return m; + } + + // Minimum age before an unparsable file is reaped, even when the TTL is 0. + // Protects a store a live process is actively creating (a sub-second window) + // from being deleted out from under it. + constexpr long kInvalidFileFloorSeconds = 2; + + // Whether an unparsable store file (bad/zero magic, wrong schema, truncated) + // is old enough to be an orphan worth removing, rather than a live process's + // store mid-creation. + [[nodiscard]] bool + invalidFileReapable(const std::filesystem::path &path, std::chrono::nanoseconds ttl) { + struct stat st{}; + if (::stat(path.c_str(), &st) != 0) { + return false; + } + const long ttl_seconds = + static_cast(std::chrono::duration_cast(ttl).count()); + const long grace = std::max(ttl_seconds, kInvalidFileFloorSeconds); + const long age = static_cast(::time(nullptr) - st.st_mtime); + return age > grace; + } + + [[nodiscard]] bool + nameMatchesStore(const std::string &name) { + return name.size() > MP_STORE_FILE_PREFIX.size() + MP_STORE_FILE_SUFFIX.size() && + name.compare(0, MP_STORE_FILE_PREFIX.size(), MP_STORE_FILE_PREFIX) == 0 && + name.compare(name.size() - MP_STORE_FILE_SUFFIX.size(), + MP_STORE_FILE_SUFFIX.size(), + MP_STORE_FILE_SUFFIX) == 0; + } + +} // namespace + +bool +isProcessAlive(int64_t pid, uint64_t start_time) { + if (pid <= 0) { + return false; + } + if (::kill(static_cast(pid), 0) != 0 && errno == ESRCH) { + return false; + } + // Process exists (kill succeeded, or failed with EPERM). Guard against PID + // reuse: if we recorded a start time and can read the current one, they must + // match. + if (start_time != 0) { + const uint64_t current = readProcessStartTime(pid); + if (current != 0 && current != start_time) { + return false; + } + } + return true; +} + +bool +isSnapshotLive(const mpStoreSnapshot &snap, std::chrono::nanoseconds ttl) { + if (isProcessAlive(snap.pid, snap.startTime)) { + return true; + } + const uint64_t now = nixlTime::getNs(); + const auto ttl_ns = static_cast(ttl.count() < 0 ? 0 : ttl.count()); + return now >= snap.lastUpdateNs && (now - snap.lastUpdateNs) <= ttl_ns; +} + +std::vector +buildMetricFamilies(const std::vector &snapshots) { + std::vector families; + if (snapshots.empty()) { + return families; + } + + for (const auto type : telemetry_metric_event_types) { + const auto descriptor = nixlEnumStrings::telemetryMetricDescriptor(type); + if (descriptor.counterName == nullptr) { + continue; + } + MetricFamily family; + family.name = descriptor.counterName; + family.help = descriptor.counterHelp; + family.type = MetricType::Counter; + const auto slot = static_cast(type); + for (const auto &snap : snapshots) { + family.metric.push_back(counterMetric(baseLabels(snap), snap.counters[slot])); + } + families.push_back(std::move(family)); + } + + for (const auto type : telemetry_metric_event_types) { + const auto descriptor = nixlEnumStrings::telemetryMetricDescriptor(type); + if (descriptor.gaugeName == nullptr) { + continue; + } + MetricFamily family; + family.name = descriptor.gaugeName; + family.help = descriptor.gaugeHelp; + family.type = MetricType::Gauge; + const auto slot = static_cast(type); + for (const auto &snap : snapshots) { + family.metric.push_back(gaugeMetric(baseLabels(snap), snap.gauges[slot])); + } + families.push_back(std::move(family)); + } + + MetricFamily errors; + errors.name = "agent_errors_total"; + errors.help = "Cumulative error count by status"; + errors.type = MetricType::Counter; + for (const auto &snap : snapshots) { + for (const auto type : telemetry_error_event_types) { + auto labels = baseLabels(snap); + labels.push_back({"status", nixlEnumStrings::telemetryErrorStatusLabel(type)}); + errors.metric.push_back( + counterMetric(std::move(labels), snap.counters[static_cast(type)])); + } + } + families.push_back(std::move(errors)); + + return families; +} + +nixlMultiprocessCollector::nixlMultiprocessCollector(std::filesystem::path dir, + std::chrono::nanoseconds stale_ttl, + bool reap_stale) + : dir_(std::move(dir)), + staleTtl_(stale_ttl), + reapStale_(reap_stale) {} + +std::vector +nixlMultiprocessCollector::Collect() const { + std::vector live; + + std::error_code ec; + std::filesystem::directory_iterator it(dir_, ec); + if (ec) { + NIXL_DEBUG << "prometheus_mp: cannot scan telemetry dir '" << dir_.string() + << "': " << ec.message(); + return {}; + } + + // Iterate with the non-throwing increment: peer writers and this collector's + // own reaping mutate the directory concurrently, and the range-for's + // operator++ would throw on a mid-iteration filesystem error. + for (const std::filesystem::directory_iterator end; it != end; it.increment(ec)) { + if (ec) { + NIXL_DEBUG << "prometheus_mp: telemetry dir iteration stopped early: " << ec.message(); + break; + } + const auto &entry = *it; + if (!entry.is_regular_file(ec) || !nameMatchesStore(entry.path().filename().string())) { + continue; + } + auto snap = readStoreSnapshot(entry.path()); + if (!snap) { + // Unparsable (mid-init, orphaned, or incompatible): reap only if it is + // old enough to not be a store a live process is currently creating. + if (reapStale_ && invalidFileReapable(entry.path(), staleTtl_)) { + std::error_code rm_ec; + std::filesystem::remove(entry.path(), rm_ec); + } + continue; + } + if (isSnapshotLive(*snap, staleTtl_)) { + live.push_back(std::move(*snap)); + } else if (reapStale_) { + std::error_code rm_ec; + std::filesystem::remove(entry.path(), rm_ec); + } + } + + return buildMetricFamilies(live); +} + +} // namespace nixl::telemetry::mp diff --git a/src/plugins/telemetry/prometheus_mp/mp_collector.h b/src/plugins/telemetry/prometheus_mp/mp_collector.h new file mode 100644 index 0000000000..b58e3b7859 --- /dev/null +++ b/src/plugins/telemetry/prometheus_mp/mp_collector.h @@ -0,0 +1,99 @@ +/* + * SPDX-FileCopyrightText: Copyright (c) 2025-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. + * SPDX-License-Identifier: Apache-2.0 + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +#ifndef NIXL_SRC_PLUGINS_TELEMETRY_PROMETHEUS_MP_MP_COLLECTOR_H +#define NIXL_SRC_PLUGINS_TELEMETRY_PROMETHEUS_MP_MP_COLLECTOR_H + +#include "mp_store.h" + +#include +#include + +#include +#include +#include + +namespace nixl::telemetry::mp { + +// Default staleness window: a store whose owning process is gone and whose last +// update is older than this is dropped (and reaped) by the collector. +inline constexpr std::chrono::seconds MP_DEFAULT_STALE_TTL{30}; + +/** + * @brief Whether a process is still running, guarding against PID reuse. + * @param pid Process id from a store header. + * @param start_time Process start time from the same header (0 = unknown/skip check). + * @return True if the pid exists and (when known) its start time still matches. + */ +[[nodiscard]] bool +isProcessAlive(int64_t pid, uint64_t start_time); + +/** + * @brief Whether a store snapshot should still be published. + * + * Live if the owning process is alive, or (to smooth over a just-exited process) + * if its last update is within @p ttl. + * @param snap Store snapshot. + * @param ttl Freshness window for a process that is no longer alive. + */ +[[nodiscard]] bool +isSnapshotLive(const mpStoreSnapshot &snap, std::chrono::nanoseconds ttl); + +/** + * @brief Converts per-process snapshots into Prometheus metric families. + * + * Emits one series per (metric, process): cumulative counters and last-operation + * gauges keyed by nixl_telemetry_event_type_t, plus the agent_errors_total family + * with a status label. Series are labeled by hostname, agent_name and (when + * present) local_rank. No cross-process aggregation -- each process is its own + * series. Returns empty when @p snapshots is empty. + * @param snapshots Live per-process snapshots. + */ +[[nodiscard]] std::vector +buildMetricFamilies(const std::vector &snapshots); + +/** + * @class nixlMultiprocessCollector + * @brief prometheus-cpp Collectable that aggregates all peer store files. + * + * Registered with the exporter process's Exposer. On each scrape it globs the + * shared directory for store files, reads a snapshot of each, drops (and + * optionally reaps) stale ones, and returns per-process metric families. + */ +class nixlMultiprocessCollector final : public prometheus::Collectable { +public: + /** + * @param dir Shared telemetry directory holding peer store files. + * @param stale_ttl Freshness window for a process that has exited. + * @param reap_stale When true, unlink store files whose process is gone and + * whose data is older than @p stale_ttl. + */ + explicit nixlMultiprocessCollector(std::filesystem::path dir, + std::chrono::nanoseconds stale_ttl = MP_DEFAULT_STALE_TTL, + bool reap_stale = true); + + [[nodiscard]] std::vector + Collect() const override; + +private: + std::filesystem::path dir_; + std::chrono::nanoseconds staleTtl_; + bool reapStale_; +}; + +} // namespace nixl::telemetry::mp + +#endif // NIXL_SRC_PLUGINS_TELEMETRY_PROMETHEUS_MP_MP_COLLECTOR_H diff --git a/src/plugins/telemetry/prometheus_mp/mp_store.cpp b/src/plugins/telemetry/prometheus_mp/mp_store.cpp new file mode 100644 index 0000000000..ac9edda590 --- /dev/null +++ b/src/plugins/telemetry/prometheus_mp/mp_store.cpp @@ -0,0 +1,275 @@ +/* + * SPDX-FileCopyrightText: Copyright (c) 2025-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. + * SPDX-License-Identifier: Apache-2.0 + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +#include "mp_store.h" + +#include "common/nixl_log.h" +#include "common/nixl_time.h" + +#include +#include +#include +#include + +#include +#include +#include +#include +#include +#include +#include +#include +#include + +namespace nixl::telemetry::mp { + +namespace { + + // "NIXLMPS1" as a little-endian tag; changing the layout must change either this + // or MP_STORE_SCHEMA_VERSION so stale-format files are rejected. + constexpr uint64_t MP_STORE_MAGIC = 0x3153504d4c58494eULL; + + constexpr std::size_t MP_MAX_AGENT_NAME = 256; + constexpr std::size_t MP_MAX_HOSTNAME = 128; + constexpr std::size_t MP_MAX_LOCAL_RANK = 64; + + // Fixed on-disk layout. Plain trivially-copyable POD operated on with __atomic + // builtins (not std::atomic) so it is safe to memset/reinterpret over an mmap'd + // region shared between processes. Field order keeps every uint64 8-byte aligned. + struct mpStoreLayout { + uint64_t magic; + uint32_t schemaVersion; + uint32_t slotCount; + int64_t pid; + uint64_t startTime; + uint64_t lastUpdateNs; + uint64_t instance; + char agentName[MP_MAX_AGENT_NAME]; + char hostname[MP_MAX_HOSTNAME]; + char localRank[MP_MAX_LOCAL_RANK]; + uint64_t counters[MP_STORE_SLOT_COUNT]; + uint64_t gauges[MP_STORE_SLOT_COUNT]; + }; + + void + copyField(char *dst, std::size_t cap, const std::string &src, const char *what) { + if (src.size() >= cap) { + NIXL_WARN << "prometheus_mp: " << what << " '" << src << "' exceeds " << (cap - 1) + << " chars; truncating in telemetry store"; + } + const std::size_t n = std::min(src.size(), cap - 1); + std::memcpy(dst, src.data(), n); + dst[n] = '\0'; + } + + [[nodiscard]] std::string + readField(const char *src, std::size_t cap) { + const std::size_t n = ::strnlen(src, cap); + return std::string(src, n); + } + +} // namespace + +std::string +makeStoreFileName(int64_t pid, uint64_t start_time, uint64_t instance) { + return std::string(MP_STORE_FILE_PREFIX) + std::to_string(pid) + "." + + std::to_string(start_time) + "." + std::to_string(instance) + + std::string(MP_STORE_FILE_SUFFIX); +} + +uint64_t +readProcessStartTime(int64_t pid) { + std::ifstream stat("/proc/" + std::to_string(pid) + "/stat"); + if (!stat.is_open()) { + return 0; + } + std::string content((std::istreambuf_iterator(stat)), std::istreambuf_iterator()); + + // comm (field 2) is wrapped in parentheses and may itself contain spaces or + // ')', so split on the LAST ')': everything after it starts at field 3. + const auto close = content.rfind(')'); + if (close == std::string::npos) { + return 0; + } + + std::istringstream rest(content.substr(close + 1)); + std::vector tokens{std::istream_iterator(rest), + std::istream_iterator()}; + // starttime is field 22; tokens[0] is field 3, so index 22 - 3 = 19. + constexpr std::size_t kStartTimeIndex = 19; + if (tokens.size() <= kStartTimeIndex) { + return 0; + } + try { + return static_cast(std::stoull(tokens[kStartTimeIndex])); + } + catch (const std::exception &) { + return 0; + } +} + +mpStoreWriter::mpStoreWriter(std::filesystem::path path, + const std::string &agent_name, + const std::string &hostname, + const std::string &local_rank, + uint64_t instance) + : path_(std::move(path)), + mappingSize_(sizeof(mpStoreLayout)) { + const int fd = ::open(path_.c_str(), O_CREAT | O_RDWR | O_CLOEXEC, 0644); + if (fd < 0) { + throw std::runtime_error("prometheus_mp: cannot open telemetry store '" + path_.string() + + "': " + std::strerror(errno)); + } + + if (::ftruncate(fd, static_cast(mappingSize_)) != 0) { + const std::string reason = std::strerror(errno); + ::close(fd); + throw std::runtime_error("prometheus_mp: cannot size telemetry store '" + path_.string() + + "': " + reason); + } + + mapping_ = ::mmap(nullptr, mappingSize_, PROT_READ | PROT_WRITE, MAP_SHARED, fd, 0); + ::close(fd); + if (mapping_ == MAP_FAILED) { + mapping_ = nullptr; + throw std::runtime_error("prometheus_mp: cannot map telemetry store '" + path_.string() + + "': " + std::strerror(errno)); + } + + auto *layout = static_cast(mapping_); + std::memset(layout, 0, mappingSize_); + layout->schemaVersion = MP_STORE_SCHEMA_VERSION; + layout->slotCount = static_cast(MP_STORE_SLOT_COUNT); + layout->pid = static_cast(::getpid()); + layout->startTime = readProcessStartTime(layout->pid); + layout->instance = instance; + copyField(layout->agentName, MP_MAX_AGENT_NAME, agent_name, "agent name"); + copyField(layout->hostname, MP_MAX_HOSTNAME, hostname, "hostname"); + copyField(layout->localRank, MP_MAX_LOCAL_RANK, local_rank, "local_rank"); + __atomic_store_n(&layout->lastUpdateNs, nixlTime::getNs(), __ATOMIC_RELEASE); + // Publish the magic last so a concurrent reader never validates a + // half-initialized header. + __atomic_store_n(&layout->magic, MP_STORE_MAGIC, __ATOMIC_RELEASE); +} + +mpStoreWriter::~mpStoreWriter() { + if (mapping_ != nullptr) { + ::munmap(mapping_, mappingSize_); + mapping_ = nullptr; + } + std::error_code ec; + std::filesystem::remove(path_, ec); +} + +void +mpStoreWriter::touch() noexcept { + auto *layout = static_cast(mapping_); + __atomic_store_n(&layout->lastUpdateNs, nixlTime::getNs(), __ATOMIC_RELEASE); +} + +void +mpStoreWriter::addCounter(nixl_telemetry_event_type_t type, uint64_t delta) noexcept { + const auto idx = static_cast(type); + if (idx >= MP_STORE_SLOT_COUNT) { + return; + } + auto *layout = static_cast(mapping_); + __atomic_fetch_add(&layout->counters[idx], delta, __ATOMIC_RELAXED); + touch(); +} + +void +mpStoreWriter::setGauge(nixl_telemetry_event_type_t type, uint64_t value) noexcept { + const auto idx = static_cast(type); + if (idx >= MP_STORE_SLOT_COUNT) { + return; + } + auto *layout = static_cast(mapping_); + __atomic_store_n(&layout->gauges[idx], value, __ATOMIC_RELAXED); + touch(); +} + +void +mpStoreWriter::refreshHeartbeat() noexcept { + touch(); +} + +std::optional +readStoreSnapshot(const std::filesystem::path &path) { + const int fd = ::open(path.c_str(), O_RDONLY | O_CLOEXEC); + if (fd < 0) { + // Missing/unreadable file is not an error here (peer may have exited). + return std::nullopt; + } + + struct stat st{}; + if (::fstat(fd, &st) != 0 || static_cast(st.st_size) < sizeof(mpStoreLayout)) { + // Too small: likely a file mid-creation by a peer. Skip quietly. + ::close(fd); + return std::nullopt; + } + + void *mapping = ::mmap(nullptr, sizeof(mpStoreLayout), PROT_READ, MAP_SHARED, fd, 0); + ::close(fd); + if (mapping == MAP_FAILED) { + NIXL_WARN << "prometheus_mp: cannot map telemetry store '" << path.string() + << "': " << std::strerror(errno); + return std::nullopt; + } + + const auto *layout = static_cast(mapping); + + const uint64_t magic = __atomic_load_n(&layout->magic, __ATOMIC_ACQUIRE); + if (magic == 0) { + // Zeroed header: either a store still being initialized by a live process, + // or an orphan left by a process that died mid-creation. Skip quietly (no + // WARN); the collector reaps stale orphans by file age. + ::munmap(mapping, sizeof(mpStoreLayout)); + return std::nullopt; + } + if (magic != MP_STORE_MAGIC) { + NIXL_WARN << "prometheus_mp: ignoring telemetry store '" << path.string() + << "' with bad magic"; + ::munmap(mapping, sizeof(mpStoreLayout)); + return std::nullopt; + } + if (layout->schemaVersion != MP_STORE_SCHEMA_VERSION || + layout->slotCount != MP_STORE_SLOT_COUNT) { + NIXL_WARN << "prometheus_mp: ignoring telemetry store '" << path.string() + << "' with incompatible schema (version " << layout->schemaVersion << ", slots " + << layout->slotCount << ")"; + ::munmap(mapping, sizeof(mpStoreLayout)); + return std::nullopt; + } + + mpStoreSnapshot snap; + snap.pid = layout->pid; + snap.startTime = layout->startTime; + snap.instance = layout->instance; + snap.lastUpdateNs = __atomic_load_n(&layout->lastUpdateNs, __ATOMIC_ACQUIRE); + snap.agentName = readField(layout->agentName, MP_MAX_AGENT_NAME); + snap.hostname = readField(layout->hostname, MP_MAX_HOSTNAME); + snap.localRank = readField(layout->localRank, MP_MAX_LOCAL_RANK); + for (std::size_t i = 0; i < MP_STORE_SLOT_COUNT; ++i) { + snap.counters[i] = __atomic_load_n(&layout->counters[i], __ATOMIC_RELAXED); + snap.gauges[i] = __atomic_load_n(&layout->gauges[i], __ATOMIC_RELAXED); + } + + ::munmap(mapping, sizeof(mpStoreLayout)); + return snap; +} + +} // namespace nixl::telemetry::mp diff --git a/src/plugins/telemetry/prometheus_mp/mp_store.h b/src/plugins/telemetry/prometheus_mp/mp_store.h new file mode 100644 index 0000000000..6c88a1580e --- /dev/null +++ b/src/plugins/telemetry/prometheus_mp/mp_store.h @@ -0,0 +1,195 @@ +/* + * SPDX-FileCopyrightText: Copyright (c) 2025-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. + * SPDX-License-Identifier: Apache-2.0 + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +#ifndef NIXL_SRC_PLUGINS_TELEMETRY_PROMETHEUS_MP_MP_STORE_H +#define NIXL_SRC_PLUGINS_TELEMETRY_PROMETHEUS_MP_MP_STORE_H + +#include "telemetry_event.h" + +#include +#include +#include +#include +#include +#include +#include + +namespace nixl::telemetry::mp { + +// On-disk schema version for the per-process metric-state store. Independent of +// the event-buffer TELEMETRY_VERSION: this is a different file format. Bump on +// any layout change so a reader rejects incompatible files. +inline constexpr uint32_t MP_STORE_SCHEMA_VERSION = 1; + +// Store file naming shared by the writer (exporter) and the collector so the +// collector can discover peer files by globbing "*". +inline constexpr std::string_view MP_STORE_FILE_PREFIX = "nixl."; +inline constexpr std::string_view MP_STORE_FILE_SUFFIX = ".mmap"; + +// Number of value slots in each of the counter and gauge arrays. Indexed +// directly by nixl_telemetry_event_type_t, so every event type has a reserved +// counter slot and a reserved gauge slot (unused ones stay zero). Derived from +// the highest enum value (AGENT_TELEMETRY_EVENTS_DROPPED must stay last); keep in +// sync if the enum is extended. +inline constexpr std::size_t MP_STORE_SLOT_COUNT = + static_cast(nixl_telemetry_event_type_t::AGENT_TELEMETRY_EVENTS_DROPPED) + 1; + +namespace detail { + // Highest slot index the collector will index into the counter/gauge arrays, + // across every event type it publishes. + [[nodiscard]] constexpr std::size_t + maxTelemetrySlot() { + std::size_t max_slot = 0; + for (const auto type : telemetry_metric_event_types) { + max_slot = std::max(max_slot, static_cast(type)); + } + for (const auto type : telemetry_error_event_types) { + max_slot = std::max(max_slot, static_cast(type)); + } + return max_slot; + } +} // namespace detail + +// Compile-time guard: if the enum is extended past AGENT_TELEMETRY_EVENTS_DROPPED +// (so it is no longer last), the fixed-slot store would be indexed out of bounds +// by the collector. Fail the build instead, forcing MP_STORE_SLOT_COUNT to be +// updated. +static_assert(detail::maxTelemetrySlot() < MP_STORE_SLOT_COUNT, + "MP_STORE_SLOT_COUNT must cover every telemetry event type the collector indexes; " + "keep AGENT_TELEMETRY_EVENTS_DROPPED last or update MP_STORE_SLOT_COUNT"); + +/** + * @brief A point-in-time copy of one process's metric-state store file. + * + * Produced by readStoreSnapshot(). Values are plain (already loaded from the + * shared file); @c counters are cumulative running totals and @c gauges hold the + * last-operation value, both indexed by nixl_telemetry_event_type_t. + */ +struct mpStoreSnapshot { + int64_t pid = 0; + // Process start time in clock ticks (/proc//stat field 22); pairs with + // pid to survive PID reuse when checking liveness. + uint64_t startTime = 0; + // Monotonic nanoseconds (nixlTime::getNs, CLOCK_MONOTONIC -- host-wide, so + // comparable across processes) of the last writer update; used for TTL staleness. + uint64_t lastUpdateNs = 0; + // Per-process instance counter, distinguishing multiple agents that share the + // same name (and thus pid/hostname) within one process so their series differ. + uint64_t instance = 0; + std::string agentName; + std::string hostname; + // Optional local (per-GPU/TP) rank label; empty when no rank env was set. + std::string localRank; + std::array counters{}; + std::array gauges{}; +}; + +/** + * @class mpStoreWriter + * @brief Owns one process's metric-state mmap file and updates it in place. + * + * Each NIXL process (writer or exporter mode) owns exactly one store. Updates + * are lock-free atomic operations directly on the mapped file, so the exporter + * process can read peers' files concurrently without coordination. The file has + * a fixed size (fixed slot layout), so it never needs to grow. + */ +class mpStoreWriter { +public: + /** + * @brief Creates (or truncates) and maps the store file at @p path. + * @param path Full path to this process's store file. + * @param agent_name Per-process agent name (unique; drives the series label). + * @param hostname Host name label. + * @param local_rank Optional rank label; pass empty to omit it. + * @param instance Per-process instance counter; disambiguates multiple + * same-named agents in one process so their series stay distinct. + * @throws std::runtime_error on open/ftruncate/mmap failure. + */ + mpStoreWriter(std::filesystem::path path, + const std::string &agent_name, + const std::string &hostname, + const std::string &local_rank, + uint64_t instance); + ~mpStoreWriter(); + + mpStoreWriter(const mpStoreWriter &) = delete; + mpStoreWriter & + operator=(const mpStoreWriter &) = delete; + mpStoreWriter(mpStoreWriter &&) = delete; + mpStoreWriter & + operator=(mpStoreWriter &&) = delete; + + // Adds @p delta to the cumulative counter slot for @p type and refreshes the + // heartbeat. Out-of-range types are ignored. + void + addCounter(nixl_telemetry_event_type_t type, uint64_t delta) noexcept; + + // Stores @p value as the last-operation gauge for @p type and refreshes the + // heartbeat. Out-of-range types are ignored. + void + setGauge(nixl_telemetry_event_type_t type, uint64_t value) noexcept; + + // Refreshes the last-update timestamp without changing any metric (keeps the + // process from looking stale during idle periods). + void + refreshHeartbeat() noexcept; + + [[nodiscard]] const std::filesystem::path & + path() const noexcept { + return path_; + } + +private: + void + touch() noexcept; + + std::filesystem::path path_; + void *mapping_ = nullptr; + std::size_t mappingSize_ = 0; +}; + +/** + * @brief Reads a consistent snapshot of a store file written by another process. + * @param path Path to a peer's store file. + * @return The snapshot, or std::nullopt if the file is missing, too small, or + * has a bad magic / incompatible schema version (a WARN is logged for a + * present-but-invalid file). + */ +[[nodiscard]] std::optional +readStoreSnapshot(const std::filesystem::path &path); + +/** + * @brief Reads a process's start time (/proc//stat field 22, clock ticks). + * @param pid Process id. + * @return The start time, or 0 if it could not be read. + */ +[[nodiscard]] uint64_t +readProcessStartTime(int64_t pid); + +/** + * @brief Builds a store file name (MP_STORE_FILE_PREFIX / suffix). + * @param pid Process id. + * @param start_time Process start time (disambiguates PID reuse across restarts). + * @param instance Per-process instance counter (disambiguates multiple agents in + * the same process so their store files never collide). + * @return File name of the form "nixl....mmap". + */ +[[nodiscard]] std::string +makeStoreFileName(int64_t pid, uint64_t start_time, uint64_t instance); + +} // namespace nixl::telemetry::mp + +#endif // NIXL_SRC_PLUGINS_TELEMETRY_PROMETHEUS_MP_MP_STORE_H diff --git a/src/plugins/telemetry/prometheus_mp/prometheus_mp_exporter.cpp b/src/plugins/telemetry/prometheus_mp/prometheus_mp_exporter.cpp new file mode 100644 index 0000000000..e109e13507 --- /dev/null +++ b/src/plugins/telemetry/prometheus_mp/prometheus_mp_exporter.cpp @@ -0,0 +1,164 @@ +/* + * SPDX-FileCopyrightText: Copyright (c) 2025-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. + * SPDX-License-Identifier: Apache-2.0 + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +#include "prometheus_mp_exporter.h" + +#include "common/configuration.h" +#include "common/nixl_log.h" + +#include + +#include +#include +#include +#include +#include +#include + +namespace { + +using nixl::telemetry::mp::makeStoreFileName; +using nixl::telemetry::mp::mpStoreWriter; +using nixl::telemetry::mp::nixlMultiprocessCollector; +using nixl::telemetry::mp::readProcessStartTime; + +constexpr uint16_t defaultPort = 9090; +constexpr uint64_t defaultStaleTtlSeconds = 30; +constexpr char defaultRankEnvName[] = "LOCAL_RANK"; + +constexpr char prometheusPortVar[] = "NIXL_TELEMETRY_PROMETHEUS_PORT"; +constexpr char prometheusLocalVar[] = "NIXL_TELEMETRY_PROMETHEUS_LOCAL"; +constexpr char multiprocDirVar[] = "NIXL_TELEMETRY_MULTIPROC_DIR"; +constexpr char rankEnvVar[] = "NIXL_TELEMETRY_RANK_ENV"; +constexpr char staleTtlVar[] = "NIXL_TELEMETRY_MP_STALE_TTL"; + +const std::string localAddress = "127.0.0.1"; +const std::string publicAddress = "0.0.0.0"; + +// civetweb reports a failed port bind with this exact text (as used by the +// single-process prometheus exporter). Only this case is treated as a benign +// bind collision; any other Exposer failure is a genuine error. +constexpr char bindFailureMarker[] = "Failed to setup server ports"; + +[[nodiscard]] std::string +getHostname() { + char hostname[HOST_NAME_MAX + 1]; + if (gethostname(hostname, sizeof(hostname)) == 0) { + hostname[HOST_NAME_MAX] = '\0'; + return std::string(hostname); + } + return "unknown"; +} + +// Resolves the optional local_rank label value: NIXL_TELEMETRY_RANK_ENV names +// which env var holds the rank (default LOCAL_RANK); the value of that env var is +// the rank. Empty when the named env var is unset -- rank is a best-effort label +// only. This is the local/per-GPU (TP) rank, distinct from Dynamo's dp_rank. +[[nodiscard]] std::string +resolveLocalRank() { + const std::string rank_source = + nixl::config::getValueDefaulted(rankEnvVar, defaultRankEnvName); + if (rank_source.empty()) { + return {}; + } + const char *value = std::getenv(rank_source.c_str()); + return value != nullptr ? std::string(value) : std::string(); +} + +[[nodiscard]] std::chrono::nanoseconds +resolveStaleTtl() { + const uint64_t seconds = + nixl::config::getValueDefaulted(staleTtlVar, defaultStaleTtlSeconds); + return std::chrono::duration_cast(std::chrono::seconds(seconds)); +} + +[[nodiscard]] std::filesystem::path +resolveMultiprocDir() { + const auto dir = nixl::config::getValueOptional(multiprocDirVar); + if (!dir || dir->empty()) { + throw std::runtime_error( + "prometheus_mp exporter requires NIXL_TELEMETRY_MULTIPROC_DIR to be set"); + } + std::filesystem::path path(*dir); + std::error_code ec; + std::filesystem::create_directories(path, ec); + if (ec) { + throw std::runtime_error("prometheus_mp exporter: cannot create telemetry dir '" + + path.string() + "': " + ec.message()); + } + return path; +} + +// Per-process instance counter so multiple agents in one process get distinct +// store files. +std::atomic s_instanceSeq{0}; + +} // namespace + +nixlTelemetryPrometheusMpExporter::nixlTelemetryPrometheusMpExporter( + const nixlTelemetryExporterInitParams &init_params) + : nixlTelemetryExporter(init_params) { + const std::filesystem::path dir = resolveMultiprocDir(); + + const int64_t pid = static_cast(::getpid()); + const uint64_t start_time = readProcessStartTime(pid); + const uint64_t instance = s_instanceSeq.fetch_add(1, std::memory_order_relaxed); + const std::filesystem::path store_path = dir / makeStoreFileName(pid, start_time, instance); + + store_ = std::make_unique( + store_path, init_params.agentName, getHostname(), resolveLocalRank(), instance); + + const bool local = nixl::config::getValueDefaulted(prometheusLocalVar, false); + const uint16_t port = nixl::config::getValueDefaulted(prometheusPortVar, defaultPort); + const std::string bind_address = + (local ? localAddress : publicAddress) + ":" + std::to_string(port); + + try { + auto exposer = std::make_shared(bind_address); + collector_ = std::make_shared(dir, resolveStaleTtl()); + exposer->RegisterCollectable(collector_); + exposer_ = std::move(exposer); + owner_ = true; + NIXL_INFO << "prometheus_mp exporter (owner) serving " << bind_address + << ", aggregating telemetry dir " << dir.string(); + } + catch (const std::exception &e) { + if (std::string(e.what()).find(bindFailureMarker) == std::string::npos) { + throw; + } + owner_ = false; + NIXL_INFO << "prometheus_mp exporter (writer): endpoint " << bind_address + << " owned by another process; agent '" << init_params.agentName + << "' writing to " << store_path.string(); + } +} + +nixlTelemetryPrometheusMpExporter::~nixlTelemetryPrometheusMpExporter() = default; + +nixl_status_t +nixlTelemetryPrometheusMpExporter::exportEvent(const nixlTelemetryEvent &event) { + const auto type = event.eventType_; + const auto descriptor = nixlEnumStrings::telemetryMetricDescriptor(type); + const bool is_error = nixlEnumStrings::telemetryErrorStatusLabel(type) != nullptr; + + if (descriptor.counterName != nullptr || is_error) { + store_->addCounter(type, event.value_); + } + if (descriptor.gaugeName != nullptr) { + store_->setGauge(type, event.value_); + } + return NIXL_SUCCESS; +} diff --git a/src/plugins/telemetry/prometheus_mp/prometheus_mp_exporter.h b/src/plugins/telemetry/prometheus_mp/prometheus_mp_exporter.h new file mode 100644 index 0000000000..919040befe --- /dev/null +++ b/src/plugins/telemetry/prometheus_mp/prometheus_mp_exporter.h @@ -0,0 +1,64 @@ +/* + * SPDX-FileCopyrightText: Copyright (c) 2025-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. + * SPDX-License-Identifier: Apache-2.0 + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +#ifndef NIXL_SRC_PLUGINS_TELEMETRY_PROMETHEUS_MP_PROMETHEUS_MP_EXPORTER_H +#define NIXL_SRC_PLUGINS_TELEMETRY_PROMETHEUS_MP_PROMETHEUS_MP_EXPORTER_H + +#include "telemetry/telemetry_exporter.h" +#include "telemetry_event.h" +#include "mp_collector.h" +#include "mp_store.h" + +#include + +#include + +/** + * @class nixlTelemetryPrometheusMpExporter + * @brief Multi-process Prometheus exporter with bind-race owner election. + * + * Every process writes its own metric state to a per-process store in the shared + * NIXL_TELEMETRY_MULTIPROC_DIR. On construction each process races to bind the + * scrape port: the winner ("owner") runs a prometheus-cpp Exposer plus a + * nixlMultiprocessCollector that aggregates all peers' stores on each scrape; the + * losers run in writer-only mode (no HTTP server). A bind collision is therefore + * benign -- it never throws nixlTelemetryBindFailed -- so every process gets a + * valid telemetry sink and all ranks are exported behind the single owner port. + */ +class nixlTelemetryPrometheusMpExporter final : public nixlTelemetryExporter { +public: + explicit nixlTelemetryPrometheusMpExporter(const nixlTelemetryExporterInitParams &init_params); + ~nixlTelemetryPrometheusMpExporter() override; + + nixl_status_t + exportEvent(const nixlTelemetryEvent &event) override; + + // True if this process won the bind race and serves the scrape endpoint. + [[nodiscard]] bool + isExporter() const noexcept { + return owner_; + } + +private: + // Declared so destruction is exposer_ -> collector_ -> store_: stop serving + // before dropping the collector it weak-references, then unmap the store. + std::unique_ptr store_; + std::shared_ptr collector_; + std::shared_ptr exposer_; + bool owner_ = false; +}; + +#endif // NIXL_SRC_PLUGINS_TELEMETRY_PROMETHEUS_MP_PROMETHEUS_MP_EXPORTER_H diff --git a/src/plugins/telemetry/prometheus_mp/prometheus_mp_plugin.cpp b/src/plugins/telemetry/prometheus_mp/prometheus_mp_plugin.cpp new file mode 100644 index 0000000000..b6ee1672be --- /dev/null +++ b/src/plugins/telemetry/prometheus_mp/prometheus_mp_plugin.cpp @@ -0,0 +1,32 @@ +/* + * SPDX-FileCopyrightText: Copyright (c) 2025-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. + * SPDX-License-Identifier: Apache-2.0 + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +#include "prometheus_mp_exporter.h" +#include "telemetry/telemetry_plugin.h" +#include "telemetry/telemetry_exporter.h" + +using prometheus_mp_exporter_plugin_t = + nixlTelemetryPluginCreator; + +extern "C" NIXL_TELEMETRY_PLUGIN_EXPORT nixlTelemetryPlugin * +nixl_telemetry_plugin_init() { + return prometheus_mp_exporter_plugin_t::create( + nixl_telemetry_plugin_api_version::V2, "prometheus_mp", "1.0.0"); +} + +extern "C" NIXL_TELEMETRY_PLUGIN_EXPORT void +nixl_telemetry_plugin_fini() {} diff --git a/test/gtest/meson.build b/test/gtest/meson.build index f0c3bddef3..d29332f0a7 100644 --- a/test/gtest/meson.build +++ b/test/gtest/meson.build @@ -89,6 +89,7 @@ gtest_sources = [ 'query_mem.cpp', 'telemetry_test.cpp', 'telemetry_prometheus_test.cpp', + 'telemetry_mp_store_test.cpp', 'telemetry_histogram_test.cpp', 'telemetry_benchmark.cpp', 'nixl_duration_test.cpp', @@ -111,11 +112,28 @@ else ucx_hw_warning_dep = [] endif +# prometheus_mp collector/exporter unit tests ride the main gtest binary, but +# only when prometheus-cpp (and thus those libraries) is available. +if is_variable('nixl_telemetry_mp_collector_dep') + gtest_sources += 'telemetry_mp_collector_test.cpp' + telemetry_mp_collector_dep = [nixl_telemetry_mp_collector_dep] +else + telemetry_mp_collector_dep = [] +endif + +if is_variable('nixl_telemetry_mp_exporter_dep') + gtest_sources += 'telemetry_mp_exporter_test.cpp' + gtest_sources += 'telemetry_mp_e2e_test.cpp' + telemetry_mp_exporter_dep = [nixl_telemetry_mp_exporter_dep] +else + telemetry_mp_exporter_dep = [] +endif + test_exe = executable('gtest', sources : gtest_sources, include_directories: [nixl_inc_dirs, utils_inc_dirs, device_api_inc, include_directories('../doca-telemetry'), include_directories('../../src/plugins/telemetry/common')], cpp_args : cpp_flags, - dependencies : [nixl_dep, nixl_common_dep, cuda_dependencies, device_api_dep, gtest_dep, gmock_dep, absl_strings_dep, absl_time_dep, ucx_hw_warning_dep, file_utils_interface], + dependencies : [nixl_dep, nixl_common_dep, cuda_dependencies, device_api_dep, gtest_dep, gmock_dep, absl_strings_dep, absl_time_dep, ucx_hw_warning_dep, file_utils_interface, nixl_telemetry_mp_store_dep, telemetry_mp_collector_dep, telemetry_mp_exporter_dep], link_with: [nixl_build_lib], install : true ) diff --git a/test/gtest/telemetry_mp_collector_test.cpp b/test/gtest/telemetry_mp_collector_test.cpp new file mode 100644 index 0000000000..108c170750 --- /dev/null +++ b/test/gtest/telemetry_mp_collector_test.cpp @@ -0,0 +1,315 @@ +/* + * SPDX-FileCopyrightText: Copyright (c) 2025-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. + * SPDX-License-Identifier: Apache-2.0 + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +#include "mp_collector.h" +#include "mp_store.h" + +#include "common/nixl_time.h" + +#include + +#include +#include +#include + +#include + +#include +#include +#include +#include +#include +#include + +namespace { + +using nixl::telemetry::mp::buildMetricFamilies; +using nixl::telemetry::mp::isProcessAlive; +using nixl::telemetry::mp::isSnapshotLive; +using nixl::telemetry::mp::makeStoreFileName; +using nixl::telemetry::mp::mpStoreSnapshot; +using nixl::telemetry::mp::mpStoreWriter; +using nixl::telemetry::mp::nixlMultiprocessCollector; +using nixl::telemetry::mp::readProcessStartTime; + +constexpr auto TX_BYTES = nixl_telemetry_event_type_t::AGENT_TX_BYTES; +constexpr auto ERR_BACKEND = nixl_telemetry_event_type_t::AGENT_ERR_BACKEND; + +[[nodiscard]] std::size_t +idx(nixl_telemetry_event_type_t t) { + return static_cast(t); +} + +[[nodiscard]] mpStoreSnapshot +makeSnap(const std::string &agent, const std::string &rank) { + mpStoreSnapshot s; + s.pid = ::getpid(); + s.startTime = 1; + s.lastUpdateNs = nixlTime::getNs(); + s.agentName = agent; + s.hostname = "host"; + s.localRank = rank; + return s; +} + +[[nodiscard]] const prometheus::MetricFamily * +findFamily(const std::vector &fams, const std::string &name) { + for (const auto &f : fams) { + if (f.name == name) { + return &f; + } + } + return nullptr; +} + +[[nodiscard]] const prometheus::ClientMetric * +findByLabel(const prometheus::MetricFamily &fam, const std::string &key, const std::string &value) { + for (const auto &m : fam.metric) { + for (const auto &l : m.label) { + if (l.name == key && l.value == value) { + return &m; + } + } + } + return nullptr; +} + +[[nodiscard]] bool +hasLabel(const prometheus::ClientMetric &m, const std::string &key) { + for (const auto &l : m.label) { + if (l.name == key) { + return true; + } + } + return false; +} + +TEST(MpCollectorTest, EmptySnapshotsYieldNoFamilies) { + EXPECT_TRUE(buildMetricFamilies({}).empty()); +} + +TEST(MpCollectorTest, PerProcessCountersAndGauges) { + auto a = makeSnap("agent-a", "0"); + a.counters[idx(TX_BYTES)] = 1000; + a.gauges[idx(TX_BYTES)] = 200; + auto b = makeSnap("agent-b", "1"); + b.counters[idx(TX_BYTES)] = 50; + b.gauges[idx(TX_BYTES)] = 50; + + const auto fams = buildMetricFamilies({a, b}); + + const auto *tx = findFamily(fams, "agent_tx_bytes_total"); + ASSERT_NE(tx, nullptr); + EXPECT_EQ(tx->type, prometheus::MetricType::Counter); + ASSERT_EQ(tx->metric.size(), 2u); + const auto *tx_a = findByLabel(*tx, "agent_name", "agent-a"); + const auto *tx_b = findByLabel(*tx, "agent_name", "agent-b"); + ASSERT_NE(tx_a, nullptr); + ASSERT_NE(tx_b, nullptr); + EXPECT_DOUBLE_EQ(tx_a->counter.value, 1000.0); + EXPECT_DOUBLE_EQ(tx_b->counter.value, 50.0); + + const auto *gauge = findFamily(fams, "agent_tx_last_bytes"); + ASSERT_NE(gauge, nullptr); + EXPECT_EQ(gauge->type, prometheus::MetricType::Gauge); + const auto *g_a = findByLabel(*gauge, "agent_name", "agent-a"); + ASSERT_NE(g_a, nullptr); + EXPECT_DOUBLE_EQ(g_a->gauge.value, 200.0); +} + +TEST(MpCollectorTest, PidLabelDisambiguatesSameAgentName) { + // Two processes that (mis)use the same agent name and no local_rank must still + // produce distinct series, keyed by pid, rather than a duplicate series. + auto a = makeSnap("dup", ""); + a.pid = 1001; + a.counters[idx(TX_BYTES)] = 10; + auto b = makeSnap("dup", ""); + b.pid = 1002; + b.counters[idx(TX_BYTES)] = 20; + + const auto fams = buildMetricFamilies({a, b}); + const auto *tx = findFamily(fams, "agent_tx_bytes_total"); + ASSERT_NE(tx, nullptr); + ASSERT_EQ(tx->metric.size(), 2u); + const auto *m_a = findByLabel(*tx, "pid", "1001"); + const auto *m_b = findByLabel(*tx, "pid", "1002"); + ASSERT_NE(m_a, nullptr); + ASSERT_NE(m_b, nullptr); + EXPECT_DOUBLE_EQ(m_a->counter.value, 10.0); + EXPECT_DOUBLE_EQ(m_b->counter.value, 20.0); + EXPECT_TRUE(hasLabel(*m_a, "pid")); +} + +TEST(MpCollectorTest, AgentInstanceLabelDisambiguatesSameProcessSameName) { + // Two agents in the SAME process (same pid) with the same name must still + // produce distinct series, keyed by agent_instance, rather than colliding. + auto a = makeSnap("dup", ""); + a.pid = 1001; + a.instance = 0; + a.counters[idx(TX_BYTES)] = 10; + auto b = makeSnap("dup", ""); + b.pid = 1001; + b.instance = 1; + b.counters[idx(TX_BYTES)] = 20; + + const auto fams = buildMetricFamilies({a, b}); + const auto *tx = findFamily(fams, "agent_tx_bytes_total"); + ASSERT_NE(tx, nullptr); + ASSERT_EQ(tx->metric.size(), 2u); + const auto *m_a = findByLabel(*tx, "agent_instance", "0"); + const auto *m_b = findByLabel(*tx, "agent_instance", "1"); + ASSERT_NE(m_a, nullptr); + ASSERT_NE(m_b, nullptr); + EXPECT_DOUBLE_EQ(m_a->counter.value, 10.0); + EXPECT_DOUBLE_EQ(m_b->counter.value, 20.0); +} + +TEST(MpCollectorTest, LocalRankLabelOnlyWhenPresent) { + const auto with_rank = makeSnap("agent-a", "3"); + const auto without_rank = makeSnap("agent-b", ""); + + const auto fams = buildMetricFamilies({with_rank, without_rank}); + const auto *tx = findFamily(fams, "agent_tx_bytes_total"); + ASSERT_NE(tx, nullptr); + + const auto *m_with = findByLabel(*tx, "agent_name", "agent-a"); + const auto *m_without = findByLabel(*tx, "agent_name", "agent-b"); + ASSERT_NE(m_with, nullptr); + ASSERT_NE(m_without, nullptr); + EXPECT_TRUE(hasLabel(*m_with, "local_rank")); + EXPECT_FALSE(hasLabel(*m_without, "local_rank")); +} + +TEST(MpCollectorTest, ErrorFamilyCarriesStatusLabel) { + auto a = makeSnap("agent-a", "0"); + a.counters[idx(ERR_BACKEND)] = 5; + + const auto fams = buildMetricFamilies({a}); + const auto *errors = findFamily(fams, "agent_errors_total"); + ASSERT_NE(errors, nullptr); + EXPECT_EQ(errors->type, prometheus::MetricType::Counter); + + const auto *backend = findByLabel(*errors, "status", "backend"); + ASSERT_NE(backend, nullptr); + EXPECT_DOUBLE_EQ(backend->counter.value, 5.0); + const auto *canceled = findByLabel(*errors, "status", "canceled"); + ASSERT_NE(canceled, nullptr); + EXPECT_DOUBLE_EQ(canceled->counter.value, 0.0); +} + +TEST(MpCollectorTest, IsProcessAliveGuardsPidReuse) { + EXPECT_TRUE(isProcessAlive(::getpid(), readProcessStartTime(::getpid()))); + EXPECT_TRUE(isProcessAlive(::getpid(), 0)); // unknown start time -> existence only + EXPECT_FALSE(isProcessAlive(0x7fffffff, 0)); // no such process + // Existing pid but wrong start time -> treated as reused -> not our process. + EXPECT_FALSE(isProcessAlive(::getpid(), 0xffffffffffffffffULL)); +} + +TEST(MpCollectorTest, SnapshotLivenessByProcessThenTtl) { + const auto ttl = std::chrono::seconds(30); + + auto alive = makeSnap("a", ""); + alive.startTime = readProcessStartTime(::getpid()); + alive.lastUpdateNs = 0; // even with an old heartbeat, a live process stays live + EXPECT_TRUE(isSnapshotLive(alive, ttl)); + + auto dead_fresh = makeSnap("b", ""); + dead_fresh.pid = 0x7fffffff; + dead_fresh.lastUpdateNs = nixlTime::getNs(); + EXPECT_TRUE(isSnapshotLive(dead_fresh, ttl)); + + auto dead_stale = makeSnap("c", ""); + dead_stale.pid = 0x7fffffff; + dead_stale.lastUpdateNs = + nixlTime::getNs() - std::chrono::nanoseconds(std::chrono::seconds(60)).count(); + EXPECT_FALSE(isSnapshotLive(dead_stale, ttl)); +} + +class MpCollectorFileTest : public ::testing::Test { +protected: + void + SetUp() override { + const auto *info = ::testing::UnitTest::GetInstance()->current_test_info(); + dir_ = std::filesystem::path(::testing::TempDir()) / + ("nixl_mp_collector_" + std::to_string(::getpid()) + "_" + info->name()); + std::filesystem::create_directories(dir_); + } + + void + TearDown() override { + std::error_code ec; + std::filesystem::remove_all(dir_, ec); + } + + std::filesystem::path dir_; +}; + +TEST_F(MpCollectorFileTest, CollectReadsLiveStoresAndIgnoresOthers) { + // Two distinct store files; both headers stamp this (live) process. + mpStoreWriter w1(dir_ / makeStoreFileName(111, 1, 0), "agent-1", "host", "0", 0); + w1.addCounter(TX_BYTES, 500); + w1.setGauge(TX_BYTES, 500); + mpStoreWriter w2(dir_ / makeStoreFileName(222, 2, 0), "agent-2", "host", "1", 0); + w2.addCounter(TX_BYTES, 700); + + // A non-store file must be ignored. + { std::ofstream(dir_ / "unrelated.txt") << "ignore me"; } + + nixlMultiprocessCollector collector(dir_, std::chrono::seconds(30), /*reap_stale=*/false); + const auto fams = collector.Collect(); + + const auto *tx = findFamily(fams, "agent_tx_bytes_total"); + ASSERT_NE(tx, nullptr); + ASSERT_EQ(tx->metric.size(), 2u); + const auto *m1 = findByLabel(*tx, "agent_name", "agent-1"); + const auto *m2 = findByLabel(*tx, "agent_name", "agent-2"); + ASSERT_NE(m1, nullptr); + ASSERT_NE(m2, nullptr); + EXPECT_DOUBLE_EQ(m1->counter.value, 500.0); + EXPECT_DOUBLE_EQ(m2->counter.value, 700.0); +} + +TEST_F(MpCollectorFileTest, ReapsOldOrphanFilesButKeepsFreshMidInitFiles) { + const auto writeZeroFile = [](const std::filesystem::path &p) { + std::ofstream f(p, std::ios::binary); + const std::string zeros(64 * 1024, '\0'); + f.write(zeros.data(), static_cast(zeros.size())); + }; + + // Orphan: zero-magic store backdated well past the reap grace -> removed. + const auto orphan = dir_ / makeStoreFileName(999, 1, 0); + writeZeroFile(orphan); + std::filesystem::last_write_time( + orphan, std::filesystem::file_time_type::clock::now() - std::chrono::hours(1)); + + // Fresh mid-init file (a live process just created it) -> must be kept. + const auto fresh = dir_ / makeStoreFileName(998, 1, 0); + writeZeroFile(fresh); + + nixlMultiprocessCollector collector(dir_, std::chrono::seconds(0), /*reap_stale=*/true); + const auto fams = collector.Collect(); + + EXPECT_TRUE(fams.empty()); + EXPECT_FALSE(std::filesystem::exists(orphan)); + EXPECT_TRUE(std::filesystem::exists(fresh)); +} + +TEST_F(MpCollectorFileTest, CollectOnEmptyDirYieldsNoFamilies) { + nixlMultiprocessCollector collector(dir_, std::chrono::seconds(30), /*reap_stale=*/false); + EXPECT_TRUE(collector.Collect().empty()); +} + +} // namespace diff --git a/test/gtest/telemetry_mp_e2e_test.cpp b/test/gtest/telemetry_mp_e2e_test.cpp new file mode 100644 index 0000000000..90bf48e0b9 --- /dev/null +++ b/test/gtest/telemetry_mp_e2e_test.cpp @@ -0,0 +1,255 @@ +/* + * SPDX-FileCopyrightText: Copyright (c) 2025-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. + * SPDX-License-Identifier: Apache-2.0 + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +#include "prometheus_mp_exporter.h" + +#include "common.h" + +#include + +#include +#include +#include +#include +#include + +#include +#include +#include +#include +#include +#include +#include +#include +#include + +namespace { + +constexpr auto TX_BYTES = nixl_telemetry_event_type_t::AGENT_TX_BYTES; + +[[nodiscard]] nixlTelemetryExporterInitParams +initParams(const std::string &agent) { + return nixlTelemetryExporterInitParams{agent, 4096}; +} + +// Minimal HTTP/1.1 GET over 127.0.0.1:; returns the response body (empty on +// failure). Self-contained to keep the test free of an HTTP client dependency. +[[nodiscard]] std::string +httpGet(uint16_t port, const std::string &path) { + const int fd = ::socket(AF_INET, SOCK_STREAM, 0); + if (fd < 0) { + return {}; + } + const struct timeval tv{3, 0}; + ::setsockopt(fd, SOL_SOCKET, SO_RCVTIMEO, &tv, sizeof(tv)); + ::setsockopt(fd, SOL_SOCKET, SO_SNDTIMEO, &tv, sizeof(tv)); + + sockaddr_in addr{}; + addr.sin_family = AF_INET; + addr.sin_port = htons(port); + addr.sin_addr.s_addr = ::inet_addr("127.0.0.1"); + if (::connect(fd, reinterpret_cast(&addr), sizeof(addr)) < 0) { + ::close(fd); + return {}; + } + + const std::string req = + "GET " + path + " HTTP/1.1\r\nHost: 127.0.0.1\r\nConnection: close\r\n\r\n"; + ::send(fd, req.data(), req.size(), 0); + + std::string response; + char buf[4096]; + while (true) { + const ssize_t n = ::recv(fd, buf, sizeof(buf), 0); + if (n <= 0) { + break; + } + response.append(buf, static_cast(n)); + } + ::close(fd); + + const auto pos = response.find("\r\n\r\n"); + return pos == std::string::npos ? std::string{} : response.substr(pos + 4); +} + +[[nodiscard]] std::string +scrapeMetrics(uint16_t port) { + const auto deadline = std::chrono::steady_clock::now() + std::chrono::seconds(3); + do { + const std::string body = httpGet(port, "/metrics"); + if (!body.empty()) { + return body; + } + std::this_thread::sleep_for(std::chrono::milliseconds(25)); + } while (std::chrono::steady_clock::now() < deadline); + return {}; +} + +// Parses "{...agent_name="X"...} " lines into agent_name -> value. +[[nodiscard]] std::map +parseSeriesByAgent(const std::string &body, const std::string &metric) { + std::map out; + const std::string prefix = metric + "{"; + const std::string key = "agent_name=\""; + std::istringstream iss(body); + std::string line; + while (std::getline(iss, line)) { + if (line.rfind(prefix, 0) != 0) { + continue; + } + const auto ap = line.find(key); + const auto brace = line.rfind('}'); + if (ap == std::string::npos || brace == std::string::npos) { + continue; + } + const auto vstart = ap + key.size(); + const auto vend = line.find('"', vstart); + if (vend == std::string::npos) { + continue; + } + try { + out[line.substr(vstart, vend - vstart)] = std::stod(line.substr(brace + 1)); + } + catch (const std::exception &) { + } + } + return out; +} + +void +runWriterChild(int go_fd, int ready_fd, int quit_fd, const std::string &agent, uint64_t tx_value) { + char c = 0; + while (::read(go_fd, &c, 1) > 0) {} + + int rc = 0; + try { + nixlTelemetryPrometheusMpExporter exporter(initParams(agent)); + exporter.exportEvent({TX_BYTES, tx_value}); + const char ok = 1; + ::write(ready_fd, &ok, 1); + // Block until the parent closes the quit pipe. + while (::read(quit_fd, &c, 1) > 0) {} + } + catch (...) { + rc = 3; + } + ::_exit(rc); +} + +class MpE2ETest : public ::testing::Test { +protected: + void + SetUp() override { + const auto *info = ::testing::UnitTest::GetInstance()->current_test_info(); + dir_ = std::filesystem::path(::testing::TempDir()) / + ("nixl_mp_e2e_" + std::to_string(::getpid()) + "_" + info->name()); + std::filesystem::create_directories(dir_); + port_ = gtest::PortAllocator::next_tcp_port(); + env_.addVar("NIXL_TELEMETRY_PROMETHEUS_LOCAL", "y"); + env_.addVar("NIXL_TELEMETRY_PROMETHEUS_PORT", std::to_string(port_)); + env_.addVar("NIXL_TELEMETRY_MULTIPROC_DIR", dir_.string()); + // Dead processes become stale immediately so the reaping check is prompt. + env_.addVar("NIXL_TELEMETRY_MP_STALE_TTL", "0"); + } + + void + TearDown() override { + for (int i = 0; i < 4; ++i) { + env_.popVar(); + } + std::error_code ec; + std::filesystem::remove_all(dir_, ec); + } + + gtest::ScopedEnv env_; + uint16_t port_ = 0; + std::filesystem::path dir_; +}; + +TEST_F(MpE2ETest, AllRankProcessesAggregateBehindOneEndpointAndStaleAreDropped) { + constexpr int kChildren = 3; + + int go_pipe[2]; + int ready_pipe[2]; + int quit_pipe[2]; + ASSERT_EQ(::pipe(go_pipe), 0); + ASSERT_EQ(::pipe(ready_pipe), 0); + ASSERT_EQ(::pipe(quit_pipe), 0); + + // Fork children while the parent is still single-threaded (before it builds + // the owner exporter, which starts civetweb threads). + std::vector children; + for (int i = 0; i < kChildren; ++i) { + const pid_t pid = ::fork(); + ASSERT_GE(pid, 0); + if (pid == 0) { + ::close(go_pipe[1]); + ::close(ready_pipe[0]); + ::close(quit_pipe[1]); + runWriterChild(go_pipe[0], + ready_pipe[1], + quit_pipe[0], + "agent-" + std::to_string(i), + static_cast((i + 1) * 100)); + } + children.push_back(pid); + } + + ::close(go_pipe[0]); + ::close(ready_pipe[1]); + ::close(quit_pipe[0]); + + // Parent becomes the bind-race owner and serves the endpoint. + nixlTelemetryPrometheusMpExporter owner(initParams("agent-parent")); + ASSERT_TRUE(owner.isExporter()); + owner.exportEvent({TX_BYTES, 999}); + + // Release the children (they now become writers) and wait for readiness. + ::close(go_pipe[1]); + for (int i = 0; i < kChildren; ++i) { + char c = 0; + ASSERT_EQ(::read(ready_pipe[0], &c, 1), 1); + } + + // Phase 1: every process must appear behind the single owner endpoint. + const auto phase1 = parseSeriesByAgent(scrapeMetrics(port_), "agent_tx_bytes_total"); + EXPECT_EQ(phase1.size(), static_cast(kChildren + 1)); + EXPECT_DOUBLE_EQ(phase1.at("agent-parent"), 999.0); + EXPECT_DOUBLE_EQ(phase1.at("agent-0"), 100.0); + EXPECT_DOUBLE_EQ(phase1.at("agent-1"), 200.0); + EXPECT_DOUBLE_EQ(phase1.at("agent-2"), 300.0); + + // Kill one child and reap it so its pid is truly gone before the next scrape. + ASSERT_EQ(::kill(children[0], SIGKILL), 0); + int status = 0; + ASSERT_EQ(::waitpid(children[0], &status, 0), children[0]); + + // Phase 2: the dead child's series is dropped (and its store reaped). + const auto phase2 = parseSeriesByAgent(scrapeMetrics(port_), "agent_tx_bytes_total"); + EXPECT_EQ(phase2.count("agent-0"), 0u); + EXPECT_EQ(phase2.count("agent-1"), 1u); + EXPECT_EQ(phase2.count("agent-2"), 1u); + EXPECT_EQ(phase2.count("agent-parent"), 1u); + + // Release the remaining children and reap them. + ::close(quit_pipe[1]); + for (int i = 1; i < kChildren; ++i) { + ::waitpid(children[i], &status, 0); + } + ::close(ready_pipe[0]); +} + +} // namespace diff --git a/test/gtest/telemetry_mp_exporter_test.cpp b/test/gtest/telemetry_mp_exporter_test.cpp new file mode 100644 index 0000000000..0fc83479b2 --- /dev/null +++ b/test/gtest/telemetry_mp_exporter_test.cpp @@ -0,0 +1,131 @@ +/* + * SPDX-FileCopyrightText: Copyright (c) 2025-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. + * SPDX-License-Identifier: Apache-2.0 + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +#include "prometheus_mp_exporter.h" +#include "mp_store.h" + +#include "common.h" + +#include + +#include + +#include + +#include +#include +#include + +namespace { + +using nixl::telemetry::mp::readStoreSnapshot; + +constexpr auto TX_BYTES = nixl_telemetry_event_type_t::AGENT_TX_BYTES; +constexpr auto RX_BYTES = nixl_telemetry_event_type_t::AGENT_RX_BYTES; + +[[nodiscard]] std::size_t +idx(nixl_telemetry_event_type_t t) { + return static_cast(t); +} + +[[nodiscard]] nixlTelemetryExporterInitParams +initParams(const std::string &agent) { + return nixlTelemetryExporterInitParams{agent, 4096}; +} + +class MpExporterTest : public ::testing::Test { +protected: + void + SetUp() override { + const auto *info = ::testing::UnitTest::GetInstance()->current_test_info(); + dir_ = std::filesystem::path(::testing::TempDir()) / + ("nixl_mp_exporter_" + std::to_string(::getpid()) + "_" + info->name()); + std::filesystem::create_directories(dir_); + port_ = gtest::PortAllocator::next_tcp_port(); + env_.addVar("NIXL_TELEMETRY_PROMETHEUS_LOCAL", "y"); + env_.addVar("NIXL_TELEMETRY_PROMETHEUS_PORT", std::to_string(port_)); + env_.addVar("NIXL_TELEMETRY_MULTIPROC_DIR", dir_.string()); + } + + void + TearDown() override { + env_.popVar(); + env_.popVar(); + env_.popVar(); + std::error_code ec; + std::filesystem::remove_all(dir_, ec); + } + + [[nodiscard]] std::filesystem::path + singleStoreFile() const { + std::error_code ec; + for (const auto &entry : std::filesystem::directory_iterator(dir_, ec)) { + const auto name = entry.path().filename().string(); + if (name.rfind("nixl.", 0) == 0 && name.size() > 5 && + name.substr(name.size() - 5) == ".mmap") { + return entry.path(); + } + } + return {}; + } + + gtest::ScopedEnv env_; + uint16_t port_ = 0; + std::filesystem::path dir_; +}; + +TEST_F(MpExporterTest, OwnerBindsAndRecordsToStore) { + nixlTelemetryPrometheusMpExporter exporter(initParams("agent-owner")); + EXPECT_TRUE(exporter.isExporter()); + + const auto file = singleStoreFile(); + ASSERT_FALSE(file.empty()); + + exporter.exportEvent({TX_BYTES, 1234}); + + const auto snap = readStoreSnapshot(file); + ASSERT_TRUE(snap.has_value()); + EXPECT_EQ(snap->agentName, "agent-owner"); + EXPECT_EQ(snap->counters[idx(TX_BYTES)], 1234u); + EXPECT_EQ(snap->gauges[idx(TX_BYTES)], 1234u); +} + +TEST_F(MpExporterTest, WriterModeWhenPortTaken) { + // Occupy the port first so the exporter loses the bind race. + prometheus::Exposer blocker("127.0.0.1:" + std::to_string(port_)); + + nixlTelemetryPrometheusMpExporter exporter(initParams("agent-writer")); + EXPECT_FALSE(exporter.isExporter()); + + const auto file = singleStoreFile(); + ASSERT_FALSE(file.empty()); + + exporter.exportEvent({RX_BYTES, 77}); + + const auto snap = readStoreSnapshot(file); + ASSERT_TRUE(snap.has_value()); + EXPECT_EQ(snap->agentName, "agent-writer"); + EXPECT_EQ(snap->counters[idx(RX_BYTES)], 77u); +} + +TEST(MpExporterStandaloneTest, MissingMultiprocDirThrows) { + ::unsetenv("NIXL_TELEMETRY_MULTIPROC_DIR"); + EXPECT_THROW( + { nixlTelemetryPrometheusMpExporter exporter(nixlTelemetryExporterInitParams{"a", 4096}); }, + std::runtime_error); +} + +} // namespace diff --git a/test/gtest/telemetry_mp_store_test.cpp b/test/gtest/telemetry_mp_store_test.cpp new file mode 100644 index 0000000000..a798b5a85f --- /dev/null +++ b/test/gtest/telemetry_mp_store_test.cpp @@ -0,0 +1,185 @@ +/* + * SPDX-FileCopyrightText: Copyright (c) 2025-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. + * SPDX-License-Identifier: Apache-2.0 + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +#include "mp_store.h" + +#include "common.h" + +#include + +#include + +#include +#include +#include +#include + +namespace { + +using nixl::telemetry::mp::mpStoreWriter; +using nixl::telemetry::mp::readProcessStartTime; +using nixl::telemetry::mp::readStoreSnapshot; + +constexpr auto TX_BYTES = nixl_telemetry_event_type_t::AGENT_TX_BYTES; +constexpr auto RX_BYTES = nixl_telemetry_event_type_t::AGENT_RX_BYTES; +constexpr auto ERR_BACKEND = nixl_telemetry_event_type_t::AGENT_ERR_BACKEND; + +[[nodiscard]] std::size_t +idx(nixl_telemetry_event_type_t t) { + return static_cast(t); +} + +class MpStoreTest : public ::testing::Test { +protected: + void + SetUp() override { + const auto *info = ::testing::UnitTest::GetInstance()->current_test_info(); + dir_ = std::filesystem::path(::testing::TempDir()) / + ("nixl_mp_store_" + std::to_string(::getpid()) + "_" + info->name()); + std::filesystem::create_directories(dir_); + } + + void + TearDown() override { + std::error_code ec; + std::filesystem::remove_all(dir_, ec); + } + + [[nodiscard]] std::filesystem::path + storePath(const std::string &name) const { + return dir_ / name; + } + + std::filesystem::path dir_; +}; + +TEST_F(MpStoreTest, WriteReadRoundTrip) { + const auto path = storePath("agent-a"); + mpStoreWriter writer(path, "agent-a", "host-1", "3", 7); + writer.addCounter(TX_BYTES, 1000); + writer.setGauge(TX_BYTES, 1000); + writer.addCounter(ERR_BACKEND, 1); + + const auto snap = readStoreSnapshot(path); + ASSERT_TRUE(snap.has_value()); + EXPECT_EQ(snap->agentName, "agent-a"); + EXPECT_EQ(snap->hostname, "host-1"); + EXPECT_EQ(snap->localRank, "3"); + EXPECT_EQ(snap->instance, 7u); + EXPECT_EQ(snap->pid, static_cast(::getpid())); + EXPECT_GT(snap->startTime, 0u); + EXPECT_GT(snap->lastUpdateNs, 0u); + EXPECT_EQ(snap->counters[idx(TX_BYTES)], 1000u); + EXPECT_EQ(snap->gauges[idx(TX_BYTES)], 1000u); + EXPECT_EQ(snap->counters[idx(ERR_BACKEND)], 1u); + // Untouched slots stay zero. + EXPECT_EQ(snap->counters[idx(RX_BYTES)], 0u); + EXPECT_EQ(snap->gauges[idx(RX_BYTES)], 0u); +} + +TEST_F(MpStoreTest, CounterAccumulatesGaugeReplaces) { + const auto path = storePath("agent-b"); + mpStoreWriter writer(path, "agent-b", "host-1", "", 0); + writer.addCounter(TX_BYTES, 100); + writer.addCounter(TX_BYTES, 250); + writer.addCounter(TX_BYTES, 650); + writer.setGauge(TX_BYTES, 100); + writer.setGauge(TX_BYTES, 650); // last write wins + + const auto snap = readStoreSnapshot(path); + ASSERT_TRUE(snap.has_value()); + EXPECT_EQ(snap->counters[idx(TX_BYTES)], 1000u); + EXPECT_EQ(snap->gauges[idx(TX_BYTES)], 650u); +} + +TEST_F(MpStoreTest, EmptyRankIsEmpty) { + const auto path = storePath("agent-c"); + mpStoreWriter writer(path, "agent-c", "host-1", "", 0); + + const auto snap = readStoreSnapshot(path); + ASSERT_TRUE(snap.has_value()); + EXPECT_TRUE(snap->localRank.empty()); +} + +TEST_F(MpStoreTest, DestructorRemovesStoreFile) { + const auto path = storePath("agent-cleanup"); + { + mpStoreWriter writer(path, "agent-cleanup", "host-1", "", 0); + EXPECT_TRUE(std::filesystem::exists(path)); + } + EXPECT_FALSE(std::filesystem::exists(path)); +} + +TEST_F(MpStoreTest, LongAgentNameTruncated) { + const auto path = storePath("agent-long"); + const std::string long_name(1000, 'x'); + const gtest::LogIgnoreGuard lig("exceeds 255 chars"); + mpStoreWriter writer(path, long_name, "host-1", "", 0); + + const auto snap = readStoreSnapshot(path); + ASSERT_TRUE(snap.has_value()); + EXPECT_EQ(snap->agentName.size(), 255u); + EXPECT_EQ(snap->agentName, long_name.substr(0, 255)); + EXPECT_EQ(lig.getIgnoredCount(), 1); +} + +TEST_F(MpStoreTest, MissingFileReturnsNullopt) { + EXPECT_FALSE(readStoreSnapshot(storePath("does-not-exist")).has_value()); +} + +TEST_F(MpStoreTest, TooSmallFileReturnsNullopt) { + const auto path = storePath("tiny"); + { + std::ofstream f(path, std::ios::binary); + const char junk[16] = {0}; + f.write(junk, sizeof(junk)); + } + EXPECT_FALSE(readStoreSnapshot(path).has_value()); +} + +TEST_F(MpStoreTest, ZeroMagicReturnsNulloptQuietly) { + const auto path = storePath("zero-magic"); + { + // Large enough to pass the size check, but all-zero: a store mid-creation + // or an orphan. Must be skipped WITHOUT a warning (no LogIgnoreGuard). + std::ofstream f(path, std::ios::binary); + const std::string zeros(64 * 1024, '\0'); + f.write(zeros.data(), static_cast(zeros.size())); + } + EXPECT_FALSE(readStoreSnapshot(path).has_value()); +} + +TEST_F(MpStoreTest, BadMagicWarnsAndReturnsNullopt) { + const auto path = storePath("bad-magic"); + { + std::ofstream f(path, std::ios::binary); + const uint64_t bad_magic = 0xDEADBEEFULL; // non-zero, not our magic + f.write(reinterpret_cast(&bad_magic), sizeof(bad_magic)); + const std::string zeros(64 * 1024, '\0'); + f.write(zeros.data(), static_cast(zeros.size())); + } + const gtest::LogIgnoreGuard lig("bad magic"); + EXPECT_FALSE(readStoreSnapshot(path).has_value()); + EXPECT_EQ(lig.getIgnoredCount(), 1); +} + +TEST_F(MpStoreTest, ProcessStartTimeSelfNonZeroBogusZero) { + EXPECT_GT(readProcessStartTime(::getpid()), 0u); + // A pid that will not exist -> 0. + EXPECT_EQ(readProcessStartTime(0x7fffffff), 0u); +} + +} // namespace