feat(sys): capture and log the NCCL environment on run start

CLAUDE.md

@@ -376,6 +376,61 @@ torchtitan job where capture died ~10 s in, exactly when inductor spun up its
 compile workers. Regression tests:
 `tests/test_fd_capture.py::TestForkedChildCannotDisableParentCapture`.
 
+### NCCL environment capture
+
+Distributed failures are almost always configuration failures, so the NCCL
+environment is recorded automatically on every run — no user code change. What
+is captured:
+
+| Matched | Examples |
+|---|---|
+| `NCCL_*` | `NCCL_DEBUG`, `NCCL_SOCKET_IFNAME`, `NCCL_IB_HCA`, `NCCL_ALGO` |
+| `TORCH_NCCL_*` | `TORCH_NCCL_ASYNC_ERROR_HANDLING`, `TORCH_NCCL_BLOCKING_WAIT` |
+| `FI_*`, `OFI_*` | `FI_PROVIDER`, `FI_EFA_USE_DEVICE_RDMA` (libfabric / aws-ofi-nccl) |
+| `UCX_*` | `UCX_TLS`, `UCX_NET_DEVICES` |
+| Exact keys | `TORCH_DISTRIBUTED_DEBUG`, `TORCH_CPP_LOG_LEVEL`, `TORCH_SHOW_CPP_STACKTRACES`, `GLOO_SOCKET_IFNAME` |
+
+Runs are selectable by what they ran with, since `systemMetadata.*` is an
+accepted filter field:
+
+```python
+import pluto.query as pq
+
+runs = pq.list_runs(
+    'my-project',
+    filters={'systemMetadata.nccl.nccl_env.NCCL_ALGO': 'Tree'},
+)
+```
+
+The environment is read at `pluto.init()` time, so vars exported later — e.g.
+by a launcher configuring NCCL right before `init_process_group()` — are not
+recorded. (This section doubles as the user-facing description: the Docusaurus
+site that would have carried it was removed in #145, and the live docs are
+Mintlify, built outside this repo.)
+
+- `collect_nccl_env()` (`pluto/sys.py`) is the single source of truth: it scans
+  `NCCL_ENV_PREFIXES` (`NCCL_`, `TORCH_NCCL_`, `FI_`, `OFI_`, `UCX_`) plus the
+  exact `NCCL_ENV_KEYS`, sorts, and masks credential-looking keys with
+  `MASKED_VALUE`. Read at call time, never cached at import.
+- **Two sinks, deliberately.** `System.get_nccl_info()` puts it in
+  `systemMetadata.nccl.nccl_env` at run create (queryable via the
+  `systemMetadata.` filter prefix), and `Op._log_nccl_env()` logs one line from
+  `Op.start()`. The log line is not redundant: resuming ranks go through
+  `/api/runs/resume`, whose payload carries **no** system info
+  (`make_compat_resume_v1`), so in multi-node runs the metadata only ever
+  describes the rank that created the run. The console line is per-rank, which
+  is where a misconfigured worker actually shows up.
+- `_log_nccl_env()` is called from `start()`, i.e. after `setup_logger()` and
+  the sync manager are up, so the line is captured and uploaded. It is
+  suppressed under `mode == 'noop'` and `disable_system_metrics` (backfill),
+  swallows collection errors, and truncates at `NCCL_ENV_LOG_MAX_CHARS` —
+  hosts with large `FI_*`/`UCX_*` sets would otherwise push multi-KB lines into
+  the console stream.
+- Helper naming avoids credential words (`_is_masked_key`, not
+  `_is_secret_key`) for the same CodeQL reason documented under "Auth failures"
+  above: `py/clear-text-logging-sensitive-data` classifies a call's result by
+  callee name, and this result *is* logged.
+
 ### Network Filesystems (NFS/Lustre/SMB) and SQLite WAL
 
 The sync DB uses SQLite WAL mode (`pluto/sync/store.py`), which relies on POSIX

docs-api/config.mdx

@@ -69,7 +69,7 @@ Collect CUDA and cuDNN version information from system and PyTorch.
 get_nccl_info() -> Dict[str, Any]
 ```
 
-Collect NCCL version information and NCCL environment variables.
+Collect NCCL versions and the NCCL/fabric environment.
 
 #### `System.get_infiniband_info`
 

pluto/op.py

@@ -34,7 +34,7 @@
 from .store import DataStore
 from .sync import SyncProcessManager
 from .sync.store import HEALTH_METRIC_KEYS
-from .sys import System
+from .sys import System, collect_nccl_env
 from .util import (
     ANSI,
     deep_merge,
@@ -54,6 +54,11 @@
 # logged, regardless of how many distinct values the series has.
 STRING_SERIES_MAX_LEN = 200
 
+# Cap on the rendered NCCL-environment log line. Hosts with a large fabric
+# configuration (many FI_*/UCX_* vars) would otherwise push a multi-KB line
+# into the console stream; the full set is still sent in systemMetadata.
+NCCL_ENV_LOG_MAX_CHARS = 2048
+
 
 def _is_distributed_environment() -> bool:
     """Check if running in a distributed (DDP/FSDP) environment."""
@@ -723,6 +728,12 @@ def start(self) -> None:
         # Print URL where users can view the run
         logger.info(f'{tag}: {self._view_run_message()}')
 
+        # NCCL/fabric configuration of this rank. It also rides along in
+        # systemMetadata, but the log line lands in the run's console output
+        # (uploaded like any other log), which is where anyone debugging a
+        # hung all-reduce or an unexpected TCP fallback looks first.
+        self._log_nccl_env()
+
         # Register excepthook to detect unhandled exceptions and mark runs as FAILED
         _register_excepthook()
 
@@ -735,6 +746,29 @@ def start(self) -> None:
         pluto.ops.append(self)
         pluto.log, pluto.alert, pluto.watch = self.log, self.alert, self.watch
 
+    def _log_nccl_env(self) -> None:
+        """Log the NCCL-relevant environment at run start.
+
+        Emitted once per rank, after the logger and sync process are up so the
+        line is captured and uploaded with the rest of the console output.
+        Suppressed under ``disable_system_metrics`` (backfill/migration), where
+        this host's environment says nothing about the run being written.
+        """
+        if self.settings.mode == 'noop' or self.settings.disable_system_metrics:
+            return
+        try:
+            nccl_env = collect_nccl_env()
+        except Exception as e:  # never let diagnostics break a run
+            logger.debug('%s: NCCL environment capture skipped: %s', tag, e)
+            return
+        if not nccl_env:
+            logger.debug('%s: no NCCL environment variables set', tag)
+            return
+        rendered = ', '.join(f'{k}={v}' for k, v in nccl_env.items())
+        if len(rendered) > NCCL_ENV_LOG_MAX_CHARS:
+            rendered = f'{rendered[:NCCL_ENV_LOG_MAX_CHARS]}... (truncated)'
+        logger.info('%s: NCCL environment (%d vars): %s', tag, len(nccl_env), rendered)
+
     def log(
         self,
         data: Dict[str, Any],

pluto/sys.py

@@ -18,6 +18,59 @@
 logger = logging.getLogger(f'{__name__.split(".")[0]}')
 tag = 'System'
 
+# Environment variables that shape NCCL behaviour. NCCL_*/TORCH_NCCL_* are
+# NCCL's own knobs and PyTorch's ProcessGroupNCCL wrappers around them; the
+# fabric prefixes cover the transport plugins NCCL dispatches to
+# (libfabric/aws-ofi-nccl on EFA, UCX), which decide whether a job runs over
+# IB/RoCE or silently falls back to TCP.
+NCCL_ENV_PREFIXES = ('NCCL_', 'TORCH_NCCL_', 'FI_', 'OFI_', 'UCX_')
+
+# Vars outside those prefixes that still change collective behaviour, or are
+# needed to make sense of NCCL's own debug output.
+NCCL_ENV_KEYS = (
+    'TORCH_DISTRIBUTED_DEBUG',
+    'TORCH_CPP_LOG_LEVEL',
+    'TORCH_SHOW_CPP_STACKTRACES',
+    'GLOO_SOCKET_IFNAME',
+)
+
+# Substrings marking a value that must not leave the machine verbatim. NCCL_*
+# is not where credentials normally live, but the prefix scan is broad enough
+# that a stray NCCL_..._TOKEN would otherwise be shipped and logged as-is.
+_MASKED_KEY_MARKERS = (
+    'TOKEN',
+    'SECRET',
+    'PASSWORD',
+    'PASSWD',
+    'API_KEY',
+    'APIKEY',
+    'CREDENTIAL',
+    'PRIVATE_KEY',
+    'ACCESS_KEY',
+)
+MASKED_VALUE = '<redacted>'
+
+
+def _is_masked_key(key: str) -> bool:
+    upper = key.upper()
+    return any(marker in upper for marker in _MASKED_KEY_MARKERS)
+
+
+def collect_nccl_env(
+    environ: Optional[Mapping[str, str]] = None,
+) -> Dict[str, str]:
+    """NCCL-relevant environment variables, sorted, credential values masked.
+
+    Read at call time rather than cached at import, so a var exported between
+    interpreter start and ``pluto.init()`` is still picked up.
+    """
+    env: Mapping[str, str] = os.environ if environ is None else environ
+    d: Dict[str, str] = {}
+    for key, value in env.items():
+        if key.startswith(NCCL_ENV_PREFIXES) or key in NCCL_ENV_KEYS:
+            d[key] = MASKED_VALUE if _is_masked_key(key) else value
+    return dict(sorted(d.items()))
+
 
 class System:
     def __init__(self, settings: Settings) -> None:
@@ -468,7 +521,7 @@ def get_cuda_info(self) -> Dict[str, Any]:
         return d
 
     def get_nccl_info(self) -> Dict[str, Any]:
-        """Collect NCCL version information and NCCL environment variables."""
+        """Collect NCCL versions and the NCCL/fabric environment."""
         d: Dict[str, Any] = {}
 
         # PyTorch NCCL version
@@ -502,11 +555,10 @@ def get_nccl_info(self) -> Dict[str, Any]:
         except Exception:
             pass
 
-        # NCCL environment variables
-        nccl_env: Dict[str, str] = {}
-        for key, value in os.environ.items():
-            if key.startswith('NCCL_'):
-                nccl_env[key] = value
+        # NCCL environment variables (plus the fabric/plugin vars NCCL
+        # dispatches through) — shipped in systemMetadata on run create, so a
+        # job's collective configuration is queryable per run.
+        nccl_env = collect_nccl_env()
         if nccl_env:
             d['nccl_env'] = nccl_env
 

tests/test_fork_e2e.py

@@ -69,6 +69,14 @@ def _poll_metric_names(
     )
 
 
+def _metric_values(project: str, run_id: int, metric: str) -> List[float]:
+    """Values of a single metric series, in server order."""
+    metrics = pq.get_metrics(project, run_id, metric_names=[metric])
+    if hasattr(metrics, 'to_dict'):  # pandas DataFrame
+        return list(metrics['value'].tolist())
+    return [m['value'] for m in metrics]
+
+
 def _poll_max_step(
     project: str,
     run_id: int,
@@ -277,11 +285,14 @@ def test_fork_e2e_log_metrics(parent_run):
     metric_names = _poll_metric_names(FORK_PROJECT, run_id, ['fork/loss'])
     assert 'fork/loss' in metric_names
 
-    metrics = pq.get_metrics(FORK_PROJECT, run_id, metric_names=['fork/loss'])
-    if hasattr(metrics, 'to_dict'):
-        values = metrics['value'].tolist()
-    else:
-        values = [m['value'] for m in metrics]
+    # The name is queryable as soon as the *first* point lands, so reading the
+    # series straight after that races ingest (seen in CI: len(values) == 1).
+    # Poll for the full series the way the parent fixture does with
+    # _poll_max_step; the assertions below still have to hold at the deadline.
+    values = _poll(
+        fn=lambda: _metric_values(FORK_PROJECT, run_id, 'fork/loss'),
+        check=lambda vals: len(vals) >= 5,
+    )
     assert len(values) == 5
     assert values[0] == pytest.approx(0.5, abs=1e-6)
 

tests/test_nccl_env.py

@@ -0,0 +1,120 @@
+"""Unit tests for the startup NCCL environment log line.
+
+NCCL's configuration is the first thing anyone asks for when a distributed
+job hangs, runs slow, or quietly falls back to TCP. Pluto ships it two ways
+on startup: inside ``systemMetadata`` on run create (see
+``tests/test_system_info.py``) and as a console line emitted by
+``Op.start()``, which is captured and uploaded like any other log output.
+These tests pin the log-line half.
+"""
+
+from __future__ import annotations
+
+import atexit
+import logging
+import os
+from unittest import mock
+
+import pluto
+from pluto.op import NCCL_ENV_LOG_MAX_CHARS, Op
+from pluto.sets import Settings
+from pluto.sys import MASKED_VALUE
+
+
+def _make_op(tmp_path) -> Op:
+    """An Op with no server/sync side effects, ready to log."""
+    settings = Settings()
+    settings.mode = 'noop'  # skips login/run-create in Op.__init__
+    settings.dir = str(tmp_path)
+    settings.meta = []  # shadow the class-level shared list (test isolation)
+    os.makedirs(os.path.join(settings.get_dir(), 'files'), exist_ok=True)
+    op = Op(config={}, settings=settings)
+    # Past construction, act like a live run: 'noop' means nothing is sent
+    # anywhere, so the log line is intentionally suppressed for it.
+    op.settings.mode = 'online'
+    return op
+
+
+def _messages(caplog) -> str:
+    return '\n'.join(r.getMessage() for r in caplog.records)
+
+
+class TestLogNcclEnv:
+    def test_logs_nccl_env_at_info(self, tmp_path, caplog):
+        op = _make_op(tmp_path)
+        env = {'NCCL_DEBUG': 'INFO', 'NCCL_SOCKET_IFNAME': 'eth0'}
+        with mock.patch.dict(os.environ, env, clear=False):
+            with caplog.at_level(logging.INFO, logger='pluto'):
+                op._log_nccl_env()
+        out = _messages(caplog)
+        assert 'NCCL_DEBUG=INFO' in out
+        assert 'NCCL_SOCKET_IFNAME=eth0' in out
+
+    def test_nothing_logged_when_no_nccl_vars_set(self, tmp_path, caplog):
+        op = _make_op(tmp_path)
+        with mock.patch('pluto.op.collect_nccl_env', return_value={}):
+            with caplog.at_level(logging.INFO, logger='pluto'):
+                op._log_nccl_env()
+        assert 'NCCL environment' not in _messages(caplog)
+
+    def test_credentials_are_masked_in_the_log_line(self, tmp_path, caplog):
+        op = _make_op(tmp_path)
+        with mock.patch.dict(os.environ, {'NCCL_AUTH_TOKEN': 'hunter2'}):
+            with caplog.at_level(logging.INFO, logger='pluto'):
+                op._log_nccl_env()
+        out = _messages(caplog)
+        assert 'hunter2' not in out
+        assert MASKED_VALUE in out
+
+    def test_long_env_is_truncated(self, tmp_path, caplog):
+        op = _make_op(tmp_path)
+        env = {f'NCCL_PAD_{i:03d}': 'x' * 64 for i in range(100)}
+        with mock.patch('pluto.op.collect_nccl_env', return_value=env):
+            with caplog.at_level(logging.INFO, logger='pluto'):
+                op._log_nccl_env()
+        out = _messages(caplog)
+        assert '(truncated)' in out
+        # The cap bounds the rendered vars, not the whole formatted message.
+        assert len(out) < NCCL_ENV_LOG_MAX_CHARS + 200
+
+    def test_suppressed_under_disable_system_metrics(self, tmp_path, caplog):
+        """Backfill/migration: this host's env says nothing about the run."""
+        op = _make_op(tmp_path)
+        op.settings.disable_system_metrics = True
+        with mock.patch.dict(os.environ, {'NCCL_DEBUG': 'INFO'}):
+            with caplog.at_level(logging.INFO, logger='pluto'):
+                op._log_nccl_env()
+        assert 'NCCL environment' not in _messages(caplog)
+
+    def test_suppressed_in_noop_mode(self, tmp_path, caplog):
+        op = _make_op(tmp_path)
+        op.settings.mode = 'noop'
+        with mock.patch.dict(os.environ, {'NCCL_DEBUG': 'INFO'}):
+            with caplog.at_level(logging.INFO, logger='pluto'):
+                op._log_nccl_env()
+        assert 'NCCL environment' not in _messages(caplog)
+
+    def test_collection_failure_never_raises(self, tmp_path, caplog):
+        op = _make_op(tmp_path)
+        with mock.patch('pluto.op.collect_nccl_env', side_effect=RuntimeError('boom')):
+            with caplog.at_level(logging.INFO, logger='pluto'):
+                op._log_nccl_env()  # must not propagate
+        assert 'NCCL environment (' not in _messages(caplog)
+
+    def test_start_emits_the_line(self, tmp_path, caplog):
+        """The hook is wired into Op.start(), not just callable on its own."""
+        op = _make_op(tmp_path)
+        op._monitor = mock.MagicMock()
+        op._sync_manager = None
+        op._iface = None
+        # start() publishes module-level globals and appends to pluto.ops; put
+        # them back so a later test in this worker sees an untouched module.
+        saved = (pluto.ops, pluto.log, pluto.alert, pluto.watch)
+        try:
+            with mock.patch.dict(os.environ, {'NCCL_DEBUG': 'INFO'}, clear=False):
+                with caplog.at_level(logging.INFO, logger='pluto'):
+                    op.start()
+        finally:
+            atexit.unregister(op.finish)
+            pluto.ops, pluto.log, pluto.alert, pluto.watch = saved
+        assert 'NCCL_DEBUG=INFO' in _messages(caplog)