`nemo_rl.telemetry.host_memory`#

Host memory per node, scraped from Ray and exported through lens.

Same source as the W&B path: every raylet publishes ray_node_mem_used and ray_node_mem_total on its own Prometheus endpoint, and RayGpuMonitorLogger in :mod:nemo_rl.utils.logger already reads them. Nothing else in NeMo-RL reports host memory – utils/memory_tracker.py samples the driver’s own RSS, prints it, and is wired only into PPO – so without this a run’s OTel stream has GPU-side numbers and no host-side ones.

Scraped here rather than teed out of the GPU monitor because that path flattens the node’s identity into the metric name (node.3.mem_gb) and discards the hostname, and because it is gated on logger.monitor_gpus: reading it would make an OTel series disappear when someone turns off a W&B feature. The cost is a second HTTP GET per node per interval, against a local endpoint, which is not measurable next to a training step.

What the number is, and is not: ray_node_mem_used is whole-host memory. It covers the raylet, the Ray object store (often tens of GB, and frequently the thing that actually ran the node out of memory), every worker resident on the host, and – on the head node – the driver. It is not a per-process or per-worker-group figure, and there is no breakdown inside it; see

data:: RL_WORKER_GROUPS_ATTR for what the worker-group attribute can and cannot be used for.

Module Contents#

Classes#

`RayNode`	One live Ray node: who it is, and where to scrape it.
`HostMemoryMonitor`	Samples every Ray node’s host memory on a timer, for lens to export.

Functions#

`_observations`	One observation per host in the latest scrape, for registry key.
`_observe_used`
`_observe_total`
`register_worker_group_nodes`	Record that worker group name_prefix occupies node_ids.
`register_worker_group_placement`	Resolve bundles to nodes and register them. Never raises.
`forget_worker_group_nodes`	Drop name_prefix’s placement, for a group that has been shut down.
`worker_group_names_on_node`	Worker groups with at least one worker on node_id, unformatted.
`format_worker_groups`	Render names as :data:`RL_WORKER_GROUPS_ATTR` carries them.
`worker_groups_on_node`	Value of :data:`RL_WORKER_GROUPS_ATTR` for node_id.
`ray_nodes`	Live Ray nodes that publish a metrics endpoint.
`parse_memory_samples`	Extract `{registry key: bytes}` from one node’s Prometheus payload.
`start_host_memory_monitoring`	Start this process’s monitor, if it has none. Never raises.
`stop_host_memory_monitoring`	Stop this process’s monitor. Never raises; a no-op if none is running.

API#

nemo_rl.telemetry.host_memory.logger#: ‘getLogger(…)’

nemo_rl.telemetry.host_memory.RL_HOST_MEMORY_USED_METRIC#: ‘rl.host.memory.used’

nemo_rl.telemetry.host_memory.RL_HOST_MEMORY_TOTAL_METRIC#: ‘rl.host.memory.total’

nemo_rl.telemetry.host_memory.HOST_NAME_ATTR#: ‘host.name’

nemo_rl.telemetry.host_memory.RL_WORKER_GROUPS_ATTR#: ‘rl.worker_groups’

nemo_rl.telemetry.host_memory.NO_WORKER_GROUPS#: ‘none’

nemo_rl.telemetry.host_memory.RAY_NODE_MEM_USED#: ‘ray_node_mem_used’

nemo_rl.telemetry.host_memory.RAY_NODE_MEM_TOTAL#: ‘ray_node_mem_total’

nemo_rl.telemetry.host_memory._SERIES_BY_RAY_NAME#: None

nemo_rl.telemetry.host_memory._USED_KEY#: ‘registry_key(…)’

nemo_rl.telemetry.host_memory._TOTAL_KEY#: ‘registry_key(…)’

nemo_rl.telemetry.host_memory._SNAPSHOT: dict[tuple[str, str], dict[str, float]]#: None

nemo_rl.telemetry.host_memory._SNAPSHOT_LOCK#: ‘Lock(…)’

nemo_rl.telemetry.host_memory._observations(key: str) → list[opentelemetry.metrics.Observation]#: One observation per host in the latest scrape, for registry key.

nemo_rl.telemetry.host_memory._observe_used( options: opentelemetry.metrics.CallbackOptions, ) → list[opentelemetry.metrics.Observation]#

nemo_rl.telemetry.host_memory._observe_total( options: opentelemetry.metrics.CallbackOptions, ) → list[opentelemetry.metrics.Observation]#

nemo_rl.telemetry.host_memory.HOST_MEMORY_RECORDED_METRICS#: ()

nemo_rl.telemetry.host_memory.FALLBACK_COLLECTION_INTERVAL#: 10.0

nemo_rl.telemetry.host_memory._SCRAPE_TIMEOUT_S#: 5.0

nemo_rl.telemetry.host_memory._SHUTDOWN_GRACE_S#: None

nemo_rl.telemetry.host_memory._WORKER_GROUP_NODES: dict[str, frozenset[str]]#: None

nemo_rl.telemetry.host_memory._WORKER_GROUP_LOCK#: ‘Lock(…)’

nemo_rl.telemetry.host_memory.register_worker_group_nodes( name_prefix: str, node_ids: Iterable[str], ) → None#

Record that worker group name_prefix occupies node_ids.

Parameters:

name_prefix – The group’s RayWorkerGroup.name_prefix – the same value its workers carry as the rl.worker_group resource attribute, so a reader can join the two.
node_ids – Ray NodeID hex strings the group’s workers are placed on.

Replaces any previous entry for the group, so a group rebuilt onto different nodes after a worker restart is not reported on both sets.

nemo_rl.telemetry.host_memory.register_worker_group_placement( name_prefix: str, bundles: Iterable[tuple[Any, int]], ) → None#

Resolve bundles to nodes and register them. Never raises.

The form :class:~nemo_rl.distributed.worker_groups.RayWorkerGroup can supply directly: it knows each worker’s (placement group, bundle index) but not the node underneath, and placement_group_table answers that from the GCS’s own scheduling record – driver-side, with no worker asked anything. Each placement group is looked up once however many bundles it contributes, since the table covers all of them.

Parameters:

name_prefix – See :func:register_worker_group_nodes.
bundles – (placement_group, bundle_index) pairs the group occupies.

nemo_rl.telemetry.host_memory.forget_worker_group_nodes(name_prefix: str) → None#: Drop name_prefix’s placement, for a group that has been shut down.

nemo_rl.telemetry.host_memory.worker_group_names_on_node(node_id: str) → set[str]#: Worker groups with at least one worker on node_id, unformatted.

nemo_rl.telemetry.host_memory.format_worker_groups(names: Iterable[str]) → str#: Render names as :data:RL_WORKER_GROUPS_ATTR carries them.

nemo_rl.telemetry.host_memory.worker_groups_on_node(node_id: str) → str#: Value of :data:RL_WORKER_GROUPS_ATTR for node_id.

class nemo_rl.telemetry.host_memory.RayNode#

One live Ray node: who it is, and where to scrape it.

node_id: str#: None

host: str#: None

metrics_url: str#: None

nemo_rl.telemetry.host_memory.ray_nodes() → list[nemo_rl.telemetry.host_memory.RayNode]#

Live Ray nodes that publish a metrics endpoint.

Returns an empty list when Ray is not up, which makes the collection loop idle rather than fail: the driver starts telemetry before init_ray().

nemo_rl.telemetry.host_memory.parse_memory_samples(metrics_text: str) → dict[str, float]#

Extract {registry key: bytes} from one node’s Prometheus payload.

Pure function (no OTel side effects) so it is trivially unit-testable. Unknown series are ignored, which is nearly all of them: a raylet publishes hundreds, and only the two in :data:_SERIES_BY_RAY_NAME are wanted.

class nemo_rl.telemetry.host_memory.HostMemoryMonitor(collection_interval: float)#

Samples every Ray node’s host memory on a timer, for lens to export.

Driver-side and read-only: it scrapes endpoints the cluster already serves and touches no worker. Start it once the cluster is up; it tolerates being started earlier and simply finds no nodes until there are some.

It publishes into :data:_SNAPSHOT rather than recording: the series are observable, so the exporter pulls the latest sample on its own schedule. The two intervals are independent, which means a sample can be up to one collection interval old when it is exported, and that samples taken between two export cycles are never seen. Both are properties of a gauge and neither is worth coupling the threads to avoid.

Initialization

Initialise the monitor.

Parameters:: collection_interval – Seconds between samples of every node. Must be positive; :func:start_host_memory_monitoring is what enforces that, since it is the only caller a run goes through.

start() → None#: Begin sampling on a background thread. Idempotent.

stop() → None#: Stop sampling and wait for the thread to finish. Idempotent.

collect_once() → None#

Sample every node and publish the result for export. Never raises.

The guarantee lives here rather than in the loop so that the loop cannot be killed by a single bad sample, and so a caller driving the monitor by hand gets the same protection.

_collect_once() → None#: Body of :meth:collect_once, inside its exception guard.

_scrape( node: nemo_rl.telemetry.host_memory.RayNode, ) → dict[str, float]#

Fetch and parse one node’s memory series, or {} if unusable.

A node that cannot be scraped is skipped rather than retried: the next interval is the retry, and a node that is down or overloaded is exactly the one that must not be allowed to stall the loop for the others.

_collection_loop() → None#: Sample until stopped, on :attr:collection_interval.

nemo_rl.telemetry.host_memory._MONITOR: Optional[nemo_rl.telemetry.host_memory.HostMemoryMonitor]#: None

nemo_rl.telemetry.host_memory.start_host_memory_monitoring(collection_interval: float) → None#

Start this process’s monitor, if it has none. Never raises.

Called from init_telemetry_driver, which is what puts the monitor in the driver process on every entry point. That is not a detail: the placement map above is an ordinary module global, so it is only visible to the process that wrote it, and RayWorkerGroup writes it from the driver. Owning the monitor anywhere else reports the right memory under the wrong label – the single-controller entry points build their worker groups on the driver (examples/run_grpo_single_controller.py, examples/run_sft_v2.py) but their Logger inside the controller actor, whose copy of the map would never be anything but empty.

Parameters:: collection_interval – Seconds between samples of every node.

nemo_rl.telemetry.host_memory.stop_host_memory_monitoring() → None#

Stop this process’s monitor. Never raises; a no-op if none is running.

Called from shutdown_telemetry, which every worker also calls – hence the no-op rather than an assertion that one is running.

`logger`
`RL_HOST_MEMORY_USED_METRIC`
`RL_HOST_MEMORY_TOTAL_METRIC`
`HOST_NAME_ATTR`
`RL_WORKER_GROUPS_ATTR`
`NO_WORKER_GROUPS`
`RAY_NODE_MEM_USED`
`RAY_NODE_MEM_TOTAL`
`_SERIES_BY_RAY_NAME`
`_USED_KEY`
`_TOTAL_KEY`
`_SNAPSHOT`
`_SNAPSHOT_LOCK`
`HOST_MEMORY_RECORDED_METRICS`
`FALLBACK_COLLECTION_INTERVAL`
`_SCRAPE_TIMEOUT_S`
`_SHUTDOWN_GRACE_S`
`_WORKER_GROUP_NODES`
`_WORKER_GROUP_LOCK`
`_MONITOR`

nemo_rl.telemetry.host_memory#

Module Contents#

Classes#

Functions#

Data#

API#

`nemo_rl.telemetry.host_memory`#