Add NCCL RAS monitoring for distributed training diagnostics

[asaiacai]

Description

This PR adds NCCL RAS (Reliability, Availability, and Serviceability) monitoring to capture diagnostic information from the NCCL RAS socket during distributed training runs.

Changes

• New module pluto/nccl_ras.py: Implements NcclRasMonitor class that:

  • Polls the local NCCL RAS socket on rank 0 only (RAS gossips OOB across all ranks)
  • Runs in a background daemon thread with configurable polling interval
  • Enqueues RAS output through the existing console-log pipeline for centralized logging
  • Handles connection failures gracefully with exponential backoff logging
  • Supports configurable RAS address via NCCL_RAS_ADDR environment variable or settings

• Modified pluto/op.py:

  • Instantiates NcclRasMonitor in Op.__init__
  • Starts RAS monitor in Op.start() alongside the system metrics monitor
  • Stops RAS monitor first in Op._teardown() to prevent enqueuing after sync store shutdown

• Modified pluto/sets.py: Added three new settings:

  • x_nccl_ras_enabled: Enable/disable RAS monitoring (default: True)
  • x_nccl_ras_addr: RAS socket address (default: '127.0.0.1:28028')
  • x_nccl_ras_log_type: Log type label for RAS output (default: 'RAS')

Design Notes

• RAS monitoring is only active on rank 0 to avoid duplicate logging across ranks
• The monitor gracefully handles cases where NCCL RAS is disabled or not listening
• Output is integrated into the existing console logging infrastructure via sync_manager.enqueue_console_batch()
• Thread lifecycle is properly managed with timeout-based join during teardown

Tests

Tested (run the relevant ones):

• Code formatting: bash format.sh
• Existing integration tests pass with RAS monitoring enabled/disabled

https://claude.ai/code/session_01QWDUivZ2pPDgNZTQ6DG9q8

[gemini-code-assist]

Code Review

This pull request introduces a new NcclRasMonitor to poll NCCL RAS data from a local socket on rank 0 and integrate it into the console-log pipeline. The monitor is managed within the Op lifecycle and includes new configuration settings. Feedback was provided to improve the robustness of the rank-zero detection logic by handling potential conversion errors and consolidating environment variable checks.

[gemini-code-assist]

The current method of checking for rank zero is not robust and can lead to an unhandled ValueError if an environment variable contains a string like '--1'. The lstrip('-').isdigit() check can pass, but int() will fail. Using a try-except block for the conversion is a safer and cleaner approach. Additionally, the logic for checking LOCAL_RANK can be merged into the main loop to avoid code duplication.

Suggested change

	MODE_NCCLRAS_JSON = 'ncclras-json'
	MODE_SOCKET_TEXT = 'socket-text'
	def _is_rank_zero() -> bool:
	"""Return True on the head process. Defaults to True when not distributed."""
	for var in ('RANK', 'SLURM_PROCID'):
	v = os.environ.get(var)
	for var in ('RANK', 'SLURM_PROCID', 'LOCAL_RANK'):
	v = os.environ.get(var)
	if v is not None:
	try:
	return int(v) == 0
	except ValueError:
	pass # Not a valid integer, try next env var
	return True