telemetry: native multi-process Prometheus aggregation (prometheus_mp)

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
 
@@ -61,28 +62,37 @@ stream:
 - **Last-operation gauges**: the value of the most recent operation, re-emitted
   unchanged (e.g. `agent_tx_last_bytes`). For example, TX byte sizes `10, 20, 35`
   yield a counter `agent_tx_bytes_total` of `65` and a gauge `agent_tx_last_bytes`
-  of `35`. The Prometheus and DOCA exporters emit **identical** series -- the same
-  names, types, and labels -- derived from one shared metric descriptor
-  (`nixlEnumStrings::telemetryMetricDescriptor` in `telemetry_event.h`). Both
+  of `35`. The Prometheus, multi-process Prometheus and DOCA exporters share
+  **identical** metric definitions -- the same names, types, and semantics --
+  derived from one shared metric descriptor
+  (`nixlEnumStrings::telemetryMetricDescriptor` in `telemetry_event.h`). The
+  emitted series are not identical, because `prometheus_mp` adds `pid`,
+  `agent_instance` and optionally `local_rank` to keep processes distinct (see
+  "Multi-process aggregation" below). All three
   expose `agent_tx_bytes_total` / `agent_rx_bytes_total` (counters, OpenMetrics
   `_total` suffix) alongside `agent_tx_last_bytes` / `agent_rx_last_bytes`
   (gauges). The memory events likewise expose both a cumulative `_total` counter
-  and a `_last_bytes` gauge on both exporters, and the transfer-time events expose
+  and a `_last_bytes` gauge on every one of them, and the transfer-time events expose
   a `_total` counter alongside a last-op gauge (`agent_xfer_time` /
   `agent_xfer_post_time`). This is purely an exporter-side derivation: no new event
   type is emitted and the buffer format is unchanged.
 - **Latency histograms**: the transfer-time events additionally feed distribution
   histograms `agent_xfer_time_us` / `agent_xfer_post_time_us` (microseconds) on
-  both the Prometheus and DOCA exporters, at parity (same names, buckets, labels).
+  the Prometheus, multi-process Prometheus and DOCA exporters, at parity (same
+  names, buckets, and semantics; `prometheus_mp` carries its extra
+  process-identity labels here too).
   Each is exposed as the standard `_bucket{le="..."}` / `_sum` / `_count` series
   alongside the existing counter and gauge. Bucket boundaries default to a
   microsecond range covering ~10us..~10s and are overridable via
   `NIXL_TELEMETRY_HISTOGRAM_BUCKETS_US` (a comma-separated list of
   strictly-increasing positive microsecond upper bounds; when absent or empty the
   built-in defaults are used, while a non-empty but invalid value is rejected and
-  the exporter fails to initialize rather than silently using the defaults). Like
-  the other views this is an exporter-side derivation with no new event type.
-- **Error counters**: the Prometheus and DOCA exporters expose error events as
+  the exporter fails to initialize rather than silently using the defaults).
+  `prometheus_mp` additionally caps the override at 32 bounds, since its buckets
+  live in a fixed-layout shared-memory store. Like the other views this is an
+  exporter-side derivation with no new event type.
+- **Error counters**: the Prometheus, multi-process Prometheus and DOCA exporters
+  expose error events as
   `agent_errors_total{status="<status>"}`. The `status` label is bounded by the
   fixed `AGENT_ERR_*` event set: `not_posted`, `invalid_param`, `backend`,
   `not_found`, `mismatch`, `not_allowed`, `repost_active`, `unknown`,
@@ -93,8 +103,8 @@ stream:
   (`updateData`, or the whole 4-event `addXferStats` batch), the event is dropped
   and counted. On each flush the core emits the count of new drops since the last
   flush as a synthetic `AGENT_TELEMETRY_EVENTS_DROPPED` event, which flows through
-  the shared descriptor as `agent_telemetry_events_dropped_total` on both
-  exporters and into the raw BUFFER stream. This counts only staging-queue loss
+  the shared descriptor as `agent_telemetry_events_dropped_total` on all three
+  metric exporters and into the raw BUFFER stream. This counts only staging-queue loss
   (events that never reach an exporter); it does **not** count BUFFER cyclic-ring
   loss from a slow downstream reader, which is a separate, uncounted condition.
 
@@ -138,7 +148,7 @@ Telemetry is configured by environment variables:
 | `NIXL_TELEMETRY_BUFFER_SIZE` | Number of events in buffer | `4096` |
 | `NIXL_TELEMETRY_RUN_INTERVAL` | Flush interval (ms) | `100` |
 | `NIXL_TELEMETRY_EXPORTER` | Name of the exporter plugin to use | - |
-| `NIXL_TELEMETRY_HISTOGRAM_BUCKETS_US` | Comma-separated microsecond bucket bounds for the transfer-time histograms (Prometheus/DOCA) | built-in µs defaults |
+| `NIXL_TELEMETRY_HISTOGRAM_BUCKETS_US` | Comma-separated microsecond bucket bounds for the transfer-time histograms (`prometheus`, `prometheus_mp`, DOCA); `prometheus_mp` accepts at most 32 bounds | built-in µs defaults |
 | `NIXL_TELEMETRY_ENABLED_METRICS` | Comma-separated allowlist of metric names to export (glob) | all |
 
 - `NIXL_TELEMETRY_ENABLE` can be set to `y`/`yes`/`on`/`true`/`enable`/`1` to be enabled, and `n`/`no`/`off`/`false`/`disable`/`0` (or not set) to be disabled. Matching is case insensitive.
@@ -147,7 +157,113 @@ 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).
+- The single-process `prometheus` exporter binds one port per process, so under multi-process runs (e.g. tensor/data parallelism) every rank tries to bind the same port and 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)). To instead export **every** rank behind one endpoint, use the `prometheus_mp` exporter, where the ranks elect one process to bind rather than racing for the port (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 for an exclusive
+`flock` on `nixl-owner.<address:port>.lock` in that directory, and only the winner binds the
+scrape port and serves `/metrics` by reading and republishing all live processes'
+files on each scrape (labeled per process). The losers run as writers only, so
+losing is benign and no rank is dropped, and a writer re-runs the election as it
+exports so the owner's death costs a gap rather than the rest of the run. If the
+winner cannot bind, nothing aggregates the directory and it says so. Should the
+lock itself be unusable, every process warns and falls back to the port bind
+deciding. 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. A departing rank leaves its store behind so its final values can still be scraped, so keep this at or above the scrape interval | `30` |
+| `NIXL_TELEMETRY_HISTOGRAM_BUCKETS_US` | Shared with the other exporters, but capped at 32 bounds by the fixed-layout store | built-in µs defaults |
+
+Unlike a bind collision, a configuration error here is fatal: a missing
+`NIXL_TELEMETRY_MULTIPROC_DIR` and an over-long
+`NIXL_TELEMETRY_HISTOGRAM_BUCKETS_US` both throw out of the `nixlAgent`
+constructor rather than leaving the rank running without telemetry.
+
+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.
+
+Whether a rank is still alive is decided by the lock it holds on its own store,
+not by its pid: a store is `flock`-ed before it is given a name and stays locked
+for the writer's lifetime, so a store the collector can lock has no writer left
+and will never change again. The kernel releases the lock however the rank died,
+which is what makes a killed rank indistinguishable from a clean exit here, and
+what lets ranks in different PID namespaces aggregate each other.
+
+As with the election, this holds while the lock is usable. A rank whose store lock
+failed warns at startup and is judged by its heartbeat alone from then on, so
+going idle for longer than the TTL costs it its series and its file -- or, on a
+filesystem where no process can lock, nothing is ever reaped and departed ranks
+stay published.
+
+### 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.
+
+The owner's death is survived rather than fatal, but not instantly: the kernel
+releases its lock, and a writer re-running the election -- which it does from its
+export path, a few times a second at most -- takes the endpoint and the reaping
+over. The endpoint is therefore unreachable for that gap plus up to one scrape
+interval. A process that exports nothing never re-elects, so a run that goes
+fully idle stays down until any rank produces telemetry again. Alert on the
+scrape target's `up` metric rather than on missing series.
+
+The owner is elected by an `flock` on `nixl-owner.<address:port>.lock` in the
+shared directory rather than by the bind itself, so exactly one process ever binds
+that address and losing the election is routine (logged at INFO). Naming the lock
+file after the address keeps it contentless and scopes the election to the ranks
+that would collide, which turns the two otherwise silent misconfigurations into
+warnings: the **owner failing to bind** means the port belongs to something outside
+the run, so nothing aggregates the directory there; and a directory **served on
+more than one address** means the ranks disagree on
+`NIXL_TELEMETRY_PROMETHEUS_PORT` (or `NIXL_TELEMETRY_PROMETHEUS_LOCAL`). Each of
+those ranks serves what it was configured with, but every one of them exports
+every rank, so scraping more than one yields the same series twice. Owners find each
+other by trying the directory's other lock files: one that can be locked is a
+leftover from an earlier run, one that cannot is a live second owner.
+
+The single-binder guarantee holds while the lock is usable. A lock file that
+cannot be opened, is not a regular file owned by the run's user, or lives on a
+filesystem without `flock` leaves every process considering itself elected, and
+the port bind decides as it did before the election existed -- still one owner
+unless the ranks also disagree on the port, in which case each binds its own.
+Every process takes that path with a warning, so it is never silent.
+
+Ranks split across directories are only detected from the abandoned side: the
+directory that did elect an owner cannot tell that ranks it never saw went
+elsewhere, so it aggregates a subset and looks healthy.
+
+For aggregation via an external service, use the DOCA/CollectX exporter instead.
 
 ## Cyclic Buffer
 

src/core/telemetry/telemetry_event.h

@@ -98,6 +98,10 @@ static_assert(nixl_telemetry_event_type_count ==
               "AGENT_TELEMETRY_EVENTS_DROPPED must remain the last enumerator; "
               "nixl_telemetry_event_type_count is out of sync with the event-type enum");
 
+// The error events share one family, so they have no per-type descriptor row.
+inline constexpr const char *telemetry_error_family_name = "agent_errors_total";
+inline constexpr const char *telemetry_error_family_help = "Cumulative error count by status";
+
 struct nixlTelemetryMetricDescriptor {
     const char *counterName;
     const char *counterHelp;
@@ -201,11 +205,12 @@ telemetryErrorStatusLabel(const nixl_telemetry_event_type_t type) noexcept {
 /**
  * @brief Exporter-side Prometheus series descriptor for a telemetry event.
  *
- * Both the native Prometheus and DOCA/CollectX exporters derive their series
- * from this single mapping, so they emit identical output. A null @c counterName,
- * @c gaugeName, or @c histogramName means the event has no cumulative counter, no
- * last-operation gauge, or no distribution histogram, respectively. Error events
- * (@c AGENT_ERR_*) and any unmapped value return an all-null descriptor.
+ * The native Prometheus, multi-process Prometheus and DOCA/CollectX exporters
+ * derive their series from this single mapping, so they emit the same metric
+ * definitions. A null @c counterName, @c gaugeName, or @c histogramName means the
+ * event has no cumulative counter, no last-operation gauge, or no distribution
+ * histogram, respectively. Error events (@c AGENT_ERR_*) and any unmapped value
+ * return an all-null descriptor.
  *
  * @param type Telemetry event type.
  * @return Counter/gauge series names and HELP strings for @p type.

src/plugins/telemetry/common/histogram_buckets.h

@@ -84,7 +84,7 @@ parseHistogramBucketsUs(const std::string &spec) {
 
 /**
  * @brief Resolve the histogram bucket boundaries (microsecond upper bounds) shared by
- *        both duration histograms in both exporters.
+ *        both duration histograms in every exporter that exposes them.
  * @return The parsed env override when set and valid; the built-in microsecond defaults
  *         when the override is absent or empty.
  * @throws std::invalid_argument when a non-empty override is provided but malformed, so a

src/plugins/telemetry/doca/doca_exporter.cpp

@@ -17,6 +17,7 @@
 #include "doca_exporter.h"
 #include "common/configuration.h"
 #include "common/exception.h"
+#include "common/hostname.h"
 #include "common/nixl_log.h"
 #include "common/str_util.h"
 #include "histogram_buckets.h"
@@ -107,16 +108,6 @@ parseExporterConfig() {
     return config;
 }
 
-[[nodiscard]] std::string
-getHostname() {
-    std::array<char, HOST_NAME_MAX + 1> hostname{};
-    if (gethostname(hostname.data(), hostname.size()) == 0) {
-        hostname.back() = '\0';
-        return std::string(hostname.data());
-    }
-    return "unknown";
-}
-
 [[nodiscard]] uint64_t
 docaTimestamp() noexcept {
     uint64_t ts = 0;
@@ -178,7 +169,7 @@ struct DocaSharedContext {
 DocaSharedContext::DocaSharedContext(const DocaExporterConfig &config)
     : scrape_enabled(config.scrape),
       ipc_enabled(config.ipc) {
-    const std::string hostname = getHostname();
+    const std::string hostname = nixl::getHostname().value_or("unknown");
     const std::string &bind_address = config.bind_address;
     const std::string &ipc_sockets_dir = config.ipc_sockets_dir;
 
@@ -395,7 +386,7 @@ nixlTelemetryDocaExporter::appendErrorCounterSample(const nixlTelemetryEvent &ev
     const char *label_values[] = {agent_name_.c_str(), status};
     return doca_telemetry_exporter_metrics_add_counter_increment(ctx_->source,
                                                                  timestamp,
-                                                                 "agent_errors_total",
+                                                                 telemetry_error_family_name,
                                                                  event.value_,
                                                                  ctx_->error_label_set_id,
                                                                  label_values);

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