From 4a9ddcc2230f9e17185918e86c67bdd8e173286c Mon Sep 17 00:00:00 2001 From: Russell Hewett Date: Fri, 28 Aug 2026 08:55:02 -0700 Subject: [PATCH 1/3] feat(trace): add explicit span emission helpers Signed-off-by: Russell Hewett --- src/nemo/lens/span_utilities.py | 99 ++++++++++++++++++++++++ tests/test_span_utilities.py | 128 ++++++++++++++++++++++++++++++++ 2 files changed, 227 insertions(+) create mode 100644 src/nemo/lens/span_utilities.py create mode 100644 tests/test_span_utilities.py diff --git a/src/nemo/lens/span_utilities.py b/src/nemo/lens/span_utilities.py new file mode 100644 index 0000000..8e7b69c --- /dev/null +++ b/src/nemo/lens/span_utilities.py @@ -0,0 +1,99 @@ +# SPDX-FileCopyrightText: Copyright (c) 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. + +"""Utilities for emitting spans from caller-supplied timestamps.""" + +from __future__ import annotations + +import os +import time +from decimal import Decimal, InvalidOperation +from pathlib import Path +from typing import Any + +from opentelemetry import trace +from opentelemetry.context import Context + +from nemo.lens.helpers import safe_set_span_attributes + +_NANOSECONDS_PER_SECOND = 1_000_000_000 + + +def emit_span( + tracer: trace.Tracer, + name: str, + start_epoch_seconds: float, + end_epoch_seconds: float, + *, + context: Context | None = None, + attributes: dict[str, Any] | None = None, +) -> trace.Span: + """Emit one span with explicit Unix-epoch second start and end times.""" + start_time = _epoch_seconds_to_nanoseconds(start_epoch_seconds, "start_epoch_seconds") + end_time = _epoch_seconds_to_nanoseconds(end_epoch_seconds, "end_epoch_seconds") + if end_time < start_time: + raise ValueError("end_epoch_seconds must be greater than or equal to start_epoch_seconds") + + span = tracer.start_span(name, context=context, start_time=start_time) + if attributes: + safe_set_span_attributes(span, attributes) + span.end(end_time=end_time) + return span + + +def linux_process_create_time( + *, + stat_text: str | None = None, + uptime_text: str | None = None, + read_time: float | None = None, + stat_path: str | os.PathLike[str] = "/proc/self/stat", + uptime_path: str | os.PathLike[str] = "/proc/uptime", + clock_ticks_per_second: int | None = None, +) -> float: + """Return process creation time in Unix-epoch seconds from Linux ``/proc`` data.""" + try: + if stat_text is None: + stat_text = Path(stat_path).read_text(encoding="utf-8") + if uptime_text is None: + uptime_text = Path(uptime_path).read_text(encoding="utf-8") + except OSError as exc: + raise RuntimeError( + "Linux process create time requires readable process stat and uptime data" + ) from exc + + if read_time is None: + read_time = time.time() + if clock_ticks_per_second is None: + clock_ticks_per_second = os.sysconf("SC_CLK_TCK") + + try: + uptime_seconds = float(uptime_text.split()[0]) + process_age_seconds = ( + int(stat_text[stat_text.rindex(")") + 2 :].split()[19]) / clock_ticks_per_second + ) + except (IndexError, TypeError, ValueError, ZeroDivisionError) as exc: + raise ValueError("Malformed Linux process stat or uptime data") from exc + + return read_time - (uptime_seconds - process_age_seconds) + + +def _epoch_seconds_to_nanoseconds(value: float, label: str) -> int: + try: + seconds = Decimal(str(value)) + except InvalidOperation as exc: + raise ValueError(f"{label} must be finite") from exc + if not seconds.is_finite(): + raise ValueError(f"{label} must be finite") + return int(seconds * _NANOSECONDS_PER_SECOND) diff --git a/tests/test_span_utilities.py b/tests/test_span_utilities.py new file mode 100644 index 0000000..a94936a --- /dev/null +++ b/tests/test_span_utilities.py @@ -0,0 +1,128 @@ +# SPDX-FileCopyrightText: Copyright (c) 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. + +"""Unit tests for span utility helpers.""" + +import pytest +from opentelemetry import trace +from opentelemetry.sdk.trace import TracerProvider +from opentelemetry.sdk.trace.export import SimpleSpanProcessor + +from nemo.lens.span_utilities import emit_span, linux_process_create_time +from tests.conftest import InMemorySpanExporter + + +@pytest.fixture +def tracer_and_exporter(): + exporter = InMemorySpanExporter() + provider = TracerProvider() + provider.add_span_processor(SimpleSpanProcessor(exporter)) + trace.set_tracer_provider(provider) + yield trace.get_tracer("test"), exporter + provider.shutdown() + + +def test_emit_span_uses_explicit_times_context_and_attributes(tracer_and_exporter): + tracer, exporter = tracer_and_exporter + parent = tracer.start_span("parent") + + emit_span( + tracer, + "test.explicit_interval", + 1_700_000_000.25, + 1_700_000_001.5, + context=trace.set_span_in_context(parent), + attributes={"phase": "startup", "ignored": None}, + ) + parent.end() + + child, _parent = exporter.get_finished_spans() + assert child.name == "test.explicit_interval" + assert child.parent.span_id == parent.context.span_id + assert child.start_time == 1_700_000_000_250_000_000 + assert child.end_time == 1_700_000_001_500_000_000 + assert child.attributes["phase"] == "startup" + assert "ignored" not in child.attributes + + +def test_emit_span_rejects_end_before_start(tracer_and_exporter): + tracer, _exporter = tracer_and_exporter + + with pytest.raises(ValueError, match="end_epoch_seconds"): + emit_span(tracer, "test.invalid", 2.0, 1.0) + + +@pytest.mark.parametrize( + ("start_epoch_seconds", "end_epoch_seconds"), + [ + (float("nan"), 1.0), + (1.0, float("inf")), + ], +) +def test_emit_span_rejects_non_finite_timestamps( + tracer_and_exporter, + start_epoch_seconds, + end_epoch_seconds, +): + tracer, _exporter = tracer_and_exporter + + with pytest.raises(ValueError, match="must be finite"): + emit_span(tracer, "test.invalid", start_epoch_seconds, end_epoch_seconds) + + +def test_linux_process_create_time_uses_start_ticks(): + stat_fields_after_comm = ["S", *(["0"] * 18), "250"] + stat_text = "123 (python worker) " + " ".join(stat_fields_after_comm) + + assert ( + linux_process_create_time( + stat_text=stat_text, + uptime_text="1000.00 2000.00", + read_time=1_700_000_000.0, + clock_ticks_per_second=100, + ) + == 1_699_999_002.5 + ) + + +@pytest.mark.parametrize( + ("stat_text", "uptime_text", "clock_ticks_per_second"), + [ + ("123 (python worker) S", "1000.00 2000.00", 100), + ("123 (python worker) " + " ".join(["S", *(["0"] * 18), "250"]), "", 100), + ("123 (python worker) " + " ".join(["S", *(["0"] * 18), "250"]), "not-a-number", 100), + ("123 (python worker) " + " ".join(["S", *(["0"] * 18), "250"]), "1000.00", 0), + ], +) +def test_linux_process_create_time_raises_for_malformed_proc_data( + stat_text, + uptime_text, + clock_ticks_per_second, +): + with pytest.raises(ValueError, match="Malformed Linux process stat or uptime data"): + linux_process_create_time( + stat_text=stat_text, + uptime_text=uptime_text, + read_time=1_700_000_000.0, + clock_ticks_per_second=clock_ticks_per_second, + ) + + +def test_linux_process_create_time_raises_when_proc_data_is_unavailable(tmp_path): + with pytest.raises(RuntimeError, match="requires readable process stat and uptime data"): + linux_process_create_time( + stat_path=tmp_path / "missing-stat", + uptime_path=tmp_path / "missing-uptime", + ) From 6f43410a22d7eda6cf8cd448a5614ab02ae5690f Mon Sep 17 00:00:00 2001 From: Russell Hewett Date: Sun, 30 Aug 2026 11:52:14 -0700 Subject: [PATCH 2/3] feat(cli): add explicit span emitter Signed-off-by: Russell Hewett --- docs/fern/versions/nightly.yml | 2 + docs/user-guide/cli.mdx | 55 +++++++ pyproject.toml | 3 + src/nemo/lens/cli.py | 249 ++++++++++++++++++++++++++++ src/nemo/lens/providers.py | 66 ++++++++ tests/test_cli.py | 287 +++++++++++++++++++++++++++++++++ 6 files changed, 662 insertions(+) create mode 100644 docs/user-guide/cli.mdx create mode 100644 src/nemo/lens/cli.py create mode 100644 tests/test_cli.py diff --git a/docs/fern/versions/nightly.yml b/docs/fern/versions/nightly.yml index 94dac3b..541d863 100644 --- a/docs/fern/versions/nightly.yml +++ b/docs/fern/versions/nightly.yml @@ -19,6 +19,8 @@ navigation: path: ../../user-guide/configuration.mdx - page: "Instrumentation Primitives" path: ../../user-guide/instrumentation.mdx + - page: "Command Line" + path: ../../user-guide/cli.mdx - page: "Span Groups" path: ../../user-guide/span-groups.mdx - page: "Metrics" diff --git a/docs/user-guide/cli.mdx b/docs/user-guide/cli.mdx new file mode 100644 index 0000000..b778a11 --- /dev/null +++ b/docs/user-guide/cli.mdx @@ -0,0 +1,55 @@ +--- +title: "Command Line" +description: "" +position: 3 +--- +NeMo Lens installs a `nemo-lens` command for launch scripts and other process +boundaries that need to emit telemetry outside a normal Python application +lifecycle. + +## `emit-spans` + +`emit-spans` emits completed spans from explicit Unix epoch-second intervals. +Use it when a shell launcher or wrapper has already measured a start and end +time and needs to export that interval as a trace span. + +Install the SDK dependencies before using the command: + +```bash +pip install 'nemo-lens[sdk]' +``` + +Pass one or more `--span` values: + +```bash +NEMO_LENS_EXPORTER=console nemo-lens emit-spans \ + --service nv.dl.launch \ + --span launch,1700000000.0,1700000003.0 \ + --span python_startup,1700000000.2,1700000001.1,launch +``` + +Each span is encoded as: + +```text +NAME,START_EPOCH_SECONDS,END_EPOCH_SECONDS[,PARENT_NAME] +``` + +Parent spans can appear before or after their children on the command line. The +command orders them before emission so the exported spans have parent context. +Parent/child intervals are not required to be nested; launch flows sometimes use +short anchor spans as parents for longer measured work. + +If any span input is invalid, nothing is emitted. Invalid input includes +malformed fields, duplicate span names, missing parents, parent cycles, +non-finite timestamps, and end times before start times. + +`--service` sets the exported `service.name`. Other attributes from +`OTEL_RESOURCE_ATTRIBUTES` are inherited by the SDK resource detector, but +`--service` takes precedence over `service.name` from the environment. + +The command uses the same exporter configuration as NeMo Lens tracing. It +defaults to OTLP; set `NEMO_LENS_EXPORTER=console` to write spans to stdout. +Standard OTel endpoint and protocol variables configure OTLP, including the +trace-specific variables, which take precedence over their generic equivalents. + +The command exits nonzero if provider setup, flushing, or span export fails. diff --git a/pyproject.toml b/pyproject.toml index 8eff6fd..3fc4ca5 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -23,6 +23,9 @@ dependencies = [ "opentelemetry-api>=1.20.0", ] +[project.scripts] +nemo-lens = "nemo.lens.cli:main" + [tool.setuptools.dynamic] version = { attr = "nemo.lens.package_info.__version__" } diff --git a/src/nemo/lens/cli.py b/src/nemo/lens/cli.py new file mode 100644 index 0000000..a773ee5 --- /dev/null +++ b/src/nemo/lens/cli.py @@ -0,0 +1,249 @@ +# SPDX-FileCopyrightText: Copyright (c) 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. + +"""Command line tools for nemo-lens.""" + +from __future__ import annotations + +import argparse +import math +import sys +from collections import Counter +from collections.abc import Sequence +from dataclasses import dataclass +from typing import Any, TextIO + +from opentelemetry import trace +from opentelemetry.context import Context + +from nemo.lens.providers import _build_span_emitter_provider +from nemo.lens.span_utilities import emit_span + + +@dataclass(frozen=True, slots=True) +class _SpanSpec: + name: str + start: float + end: float + parent: str | None = None + + +@dataclass(frozen=True, slots=True) +class _SpanValidation: + order: list[_SpanSpec] + invalid_count: int = 0 + + +def main(argv: Sequence[str] | None = None) -> int: + """Run the ``nemo-lens`` command line interface.""" + parser = _build_parser() + args = parser.parse_args(argv) + + if args.command == "emit-spans": + return _run_emit_spans(args) + + parser.error("missing command") + return 2 + + +def _build_parser() -> argparse.ArgumentParser: + parser = argparse.ArgumentParser(prog="nemo-lens") + subparsers = parser.add_subparsers(dest="command", required=True) + + emit_spans = subparsers.add_parser( + "emit-spans", + description="Emit spans from explicit epoch-second intervals.", + help="emit spans from explicit intervals", + ) + emit_spans.add_argument("--service", required=True, help="OpenTelemetry service.name") + emit_spans.add_argument( + "--span", + action="append", + default=[], + required=True, + metavar="NAME,START,END[,PARENT]", + help="Span name, start epoch seconds, end epoch seconds, and optional parent name.", + ) + return parser + + +def _run_emit_spans( + args: argparse.Namespace, + *, + span_exporter: Any | None = None, + stderr: TextIO | None = None, +) -> int: + stderr = stderr or sys.stderr + validation = _validate_span_args(args.span, stderr=stderr) + if validation.invalid_count or not validation.order: + print("nothing emitted: invalid span input", file=stderr) + return 1 + order = validation.order + + try: + provider, tracked_exporter = _build_span_emitter_provider( + args.service, + span_exporter=span_exporter, + ) + except (ImportError, ValueError) as exc: + print(f"error: {exc}", file=stderr) + return 2 + + tracer = trace.get_tracer(args.service) + emitted: dict[str, trace.Span] = {} + flush_succeeded = False + try: + for spec in order: + context = trace.set_span_in_context(emitted[spec.parent]) if spec.parent else Context() + emitted[spec.name] = emit_span( + tracer, + spec.name, + spec.start, + spec.end, + context=context, + ) + + flush_succeeded = provider.force_flush() + finally: + provider.shutdown() + + if not flush_succeeded: + print("error: timed out while flushing spans", file=stderr) + return 2 + if tracked_exporter.failed: + print("error: failed to export spans", file=stderr) + return 2 + + for spec in order: + print( + f" {spec.name:34s} {spec.end - spec.start:8.3f}s" + f"{'' if spec.parent is None else ' -> ' + spec.parent}", + file=stderr, + ) + return 0 + + +def _validate_span_args( + span_args: Sequence[str], *, stderr: TextIO | None = None +) -> _SpanValidation: + stderr = stderr or sys.stderr + spans: list[_SpanSpec] = [] + invalid_count = 0 + + for span_arg in span_args: + spec = _parse_span_arg(span_arg, stderr=stderr) + if spec is None: + invalid_count += 1 + continue + spans.append(spec) + + ordered = _order_span_specs(spans, stderr=stderr) + return _SpanValidation(ordered.order, invalid_count + ordered.invalid_count) + + +def _parse_span_arg(arg: str, *, stderr: TextIO | None = None) -> _SpanSpec | None: + stderr = stderr or sys.stderr + parts = [part.strip() for part in arg.split(",")] + if len(parts) not in (3, 4): + print(f"error: skipping {arg!r}: expected NAME,START,END[,PARENT]", file=stderr) + return None + + name = parts[0] + start = parts[1] + end = parts[2] + parent = parts[3] if len(parts) == 4 else "" + + if not name: + print(f"error: skipping {arg!r}: missing span name", file=stderr) + return None + + if not start or not end: + print(f"error: skipping {name!r}: missing timestamp", file=stderr) + return None + + try: + start_epoch_seconds = float(start) + end_epoch_seconds = float(end) + except ValueError: + print(f"error: skipping {name!r}: unparseable timestamp", file=stderr) + return None + + if not math.isfinite(start_epoch_seconds) or not math.isfinite(end_epoch_seconds): + print(f"error: skipping {name!r}: timestamp must be finite", file=stderr) + return None + + if end_epoch_seconds < start_epoch_seconds: + print(f"error: skipping {name!r}: end timestamp precedes start timestamp", file=stderr) + return None + + return _SpanSpec(name, start_epoch_seconds, end_epoch_seconds, parent or None) + + +def _order_span_specs( + spans: Sequence[_SpanSpec], *, stderr: TextIO | None = None +) -> _SpanValidation: + stderr = stderr or sys.stderr + invalid_count = 0 + + name_counts = Counter(spec.name for spec in spans) + duplicate_names = {name for name, count in name_counts.items() if count > 1} + for name in sorted(duplicate_names): + print(f"error: invalid span {name!r}: duplicate span name", file=stderr) + invalid_count += 1 + + candidate_specs: list[_SpanSpec] = [] + candidate_names = {spec.name for spec in spans if spec.name not in duplicate_names} + for spec in spans: + if spec.name in duplicate_names: + continue + if spec.parent in duplicate_names: + print( + f"error: invalid span {spec.name!r}: parent {spec.parent!r} is duplicated", + file=stderr, + ) + invalid_count += 1 + elif spec.parent is not None and spec.parent not in candidate_names: + print(f"error: invalid span {spec.name!r}: missing parent {spec.parent!r}", file=stderr) + invalid_count += 1 + else: + candidate_specs.append(spec) + + if invalid_count: + return _SpanValidation([], invalid_count) + + order: list[_SpanSpec] = [] + ordered_names: set[str] = set() + remaining = list(candidate_specs) + + while remaining: + ready = [spec for spec in remaining if spec.parent is None or spec.parent in ordered_names] + if not ready: + for spec in remaining: + print( + f"error: invalid span graph: parent cycle prevents emitting {spec.name!r}", + file=stderr, + ) + invalid_count += 1 + return _SpanValidation([], invalid_count) + + order.extend(ready) + ordered_names.update(spec.name for spec in ready) + remaining = [spec for spec in remaining if spec.name not in ordered_names] + + return _SpanValidation(order) + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/src/nemo/lens/providers.py b/src/nemo/lens/providers.py index d31d9c0..c9743f0 100644 --- a/src/nemo/lens/providers.py +++ b/src/nemo/lens/providers.py @@ -158,6 +158,35 @@ def shutdown(self) -> None: logging.getLogger(__name__).debug("Failed to end span on shutdown", exc_info=True) +class _ExportResultTrackingSpanExporter: + """Delegate span export while retaining failures discarded by the batch processor.""" + + def __init__(self, exporter, success_result) -> None: + self._exporter = exporter + self._success_result = success_result + self._failed = threading.Event() + + @property + def failed(self) -> bool: + return self._failed.is_set() + + def export(self, spans): + try: + result = self._exporter.export(spans) + except Exception: + self._failed.set() + raise + if result != self._success_result: + self._failed.set() + return result + + def shutdown(self) -> None: + self._exporter.shutdown() + + def force_flush(self, timeout_millis: int = 30000) -> bool: + return self._exporter.force_flush(timeout_millis) + + class SeedIndependentIdGenerator: """OTel-compatible IdGenerator whose IDs survive a global ``random.seed()``. @@ -327,6 +356,43 @@ def build_noop_providers() -> None: metrics.set_meter_provider(NoOpMeterProvider()) +def _build_span_emitter_provider(service_name: str, span_exporter=None): + """Build the trace provider used by ``nemo-lens emit-spans``.""" + try: + from opentelemetry import trace + from opentelemetry.sdk.resources import Resource + from opentelemetry.sdk.trace import TracerProvider + from opentelemetry.sdk.trace.export import BatchSpanProcessor, SpanExportResult + except ImportError as exc: + raise ImportError( + "nemo-lens emit-spans requires OpenTelemetry SDK dependencies. " + "Install with: pip install 'nemo-lens[sdk]'" + ) from exc + + if span_exporter is None: + from nemo.lens.config import NemoLensConfig + + span_exporter = _build_span_exporter(NemoLensConfig.from_env()) + + tracked_exporter = _ExportResultTrackingSpanExporter( + span_exporter, + SpanExportResult.SUCCESS, + ) + + from nemo.lens.resources import detect_resource + + resource_attributes = detect_resource() + resource_attributes["service.name"] = service_name + + provider = TracerProvider( + resource=Resource.create(resource_attributes), + id_generator=SeedIndependentIdGenerator(), + ) + provider.add_span_processor(BatchSpanProcessor(tracked_exporter)) + trace.set_tracer_provider(provider) + return provider, tracked_exporter + + # --------------------------------------------------------------------------- # Internal helpers # --------------------------------------------------------------------------- diff --git a/tests/test_cli.py b/tests/test_cli.py new file mode 100644 index 0000000..506a28c --- /dev/null +++ b/tests/test_cli.py @@ -0,0 +1,287 @@ +# SPDX-FileCopyrightText: Copyright (c) 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. + +"""Unit tests for the nemo-lens command line interface.""" + +import io + +import pytest +from opentelemetry.sdk.trace.export import SpanExportResult + +from nemo.lens import cli, providers +from tests.conftest import InMemorySpanExporter + + +class _FailingSpanExporter(InMemorySpanExporter): + def export(self, spans): + return SpanExportResult.FAILURE + + +class _RaisingSpanExporter(InMemorySpanExporter): + def export(self, spans): + raise RuntimeError("export failed") + + +def test_parse_span_arg_accepts_optional_parent(): + assert cli._parse_span_arg("child,1.25,2.5,parent") == cli._SpanSpec( + "child", 1.25, 2.5, "parent" + ) + + +def test_parse_span_arg_warns_for_missing_and_unparseable_timestamps(): + stderr = io.StringIO() + + assert cli._parse_span_arg("missing-start,,2", stderr=stderr) is None + assert cli._parse_span_arg("bad,one,2", stderr=stderr) is None + assert cli._parse_span_arg(",1,2", stderr=stderr) is None + + warnings = stderr.getvalue() + assert "error: skipping 'missing-start': missing timestamp" in warnings + assert "error: skipping 'bad': unparseable timestamp" in warnings + assert "error: skipping ',1,2': missing span name" in warnings + + +def test_parse_span_arg_rejects_end_before_start(): + stderr = io.StringIO() + + assert cli._parse_span_arg("backwards,2,1", stderr=stderr) is None + + assert ( + "error: skipping 'backwards': end timestamp precedes start timestamp" in stderr.getvalue() + ) + + +def test_order_span_specs_emits_parents_before_children(): + validation = cli._order_span_specs( + [ + cli._SpanSpec("child", 2.0, 3.0, "root"), + cli._SpanSpec("root", 1.0, 4.0), + cli._SpanSpec("grandchild", 2.5, 2.75, "child"), + ] + ) + + assert validation.invalid_count == 0 + assert [spec.name for spec in validation.order] == ["root", "child", "grandchild"] + + +def test_order_span_specs_rejects_duplicate_or_missing_parent(): + stderr = io.StringIO() + + validation = cli._order_span_specs( + [ + cli._SpanSpec("root", 1.0, 4.0), + cli._SpanSpec("root", 2.0, 3.0), + cli._SpanSpec("orphan", 1.0, 2.0, "missing"), + ], + stderr=stderr, + ) + + assert validation.order == [] + assert validation.invalid_count == 2 + messages = stderr.getvalue() + assert "error: invalid span 'root': duplicate span name" in messages + assert "error: invalid span 'orphan': missing parent 'missing'" in messages + + +def test_main_emit_spans_uses_explicit_times_parent_context_and_env_resource(monkeypatch): + exporter = InMemorySpanExporter() + monkeypatch.setenv( + "OTEL_RESOURCE_ATTRIBUTES", + ( + "service.name=from-env,slurm.job.id=123,slurm.array.count=1," + "slurm.nnodes=2,slurm.ntasks=2,slurm.restart_count=0,cluster.name=atlas" + ), + ) + args = cli._build_parser().parse_args( + [ + "emit-spans", + "--service", + "nemo-test", + "--span", + "child,1700000001.5,1700000002.25,parent", + "--span", + "parent,1700000000,1700000003", + ] + ) + + assert cli._run_emit_spans(args, span_exporter=exporter) == 0 + + spans = {span.name: span for span in exporter.get_finished_spans()} + child = spans["child"] + parent = spans["parent"] + + assert child.parent.span_id == parent.context.span_id + assert child.start_time == 1_700_000_001_500_000_000 + assert child.end_time == 1_700_000_002_250_000_000 + assert parent.resource.attributes["service.name"] == "nemo-test" + assert parent.resource.attributes["slurm.job.id"] == "123" + assert isinstance(parent.resource.attributes["slurm.job.id"], str) + assert parent.resource.attributes["slurm.array.count"] == 1 + assert isinstance(parent.resource.attributes["slurm.array.count"], int) + assert parent.resource.attributes["slurm.nnodes"] == 2 + assert isinstance(parent.resource.attributes["slurm.nnodes"], int) + assert parent.resource.attributes["slurm.ntasks"] == 2 + assert isinstance(parent.resource.attributes["slurm.ntasks"], int) + assert parent.resource.attributes["slurm.restart_count"] == 0 + assert isinstance(parent.resource.attributes["slurm.restart_count"], int) + assert parent.resource.attributes["cluster.name"] == "atlas" + + +def test_main_emit_spans_service_overrides_env_service_name(monkeypatch): + exporter = InMemorySpanExporter() + monkeypatch.setenv("OTEL_RESOURCE_ATTRIBUTES", "service.name=from-env,cluster.name=atlas") + args = cli._build_parser().parse_args( + [ + "emit-spans", + "--service", + "from-cli", + "--span", + "root,1700000000,1700000001", + ] + ) + + assert cli._run_emit_spans(args, span_exporter=exporter) == 0 + + (span,) = exporter.get_finished_spans() + assert span.resource.attributes["service.name"] == "from-cli" + assert span.resource.attributes["cluster.name"] == "atlas" + + +def test_main_emit_spans_returns_error_when_provider_setup_fails(monkeypatch): + stderr = io.StringIO() + + def fail_provider_setup(*args, **kwargs): + raise ImportError("missing OpenTelemetry SDK") + + monkeypatch.setattr(cli, "_build_span_emitter_provider", fail_provider_setup) + args = cli._build_parser().parse_args( + [ + "emit-spans", + "--service", + "nemo-test", + "--span", + "root,1700000000,1700000001", + ] + ) + + assert cli._run_emit_spans(args, stderr=stderr) == 2 + assert "error: missing OpenTelemetry SDK" in stderr.getvalue() + + +@pytest.mark.parametrize( + ("configured_exporter", "expected_exporter"), + [(None, "otlp"), ("console", "console")], +) +def test_main_emit_spans_uses_lens_exporter_config( + monkeypatch, + configured_exporter, + expected_exporter, +): + selected = [] + span_exporter = InMemorySpanExporter() + if configured_exporter is None: + monkeypatch.delenv("NEMO_LENS_EXPORTER", raising=False) + else: + monkeypatch.setenv("NEMO_LENS_EXPORTER", configured_exporter) + monkeypatch.setattr( + providers, + "_build_span_exporter", + lambda config: selected.append(config.exporter) or span_exporter, + ) + args = cli._build_parser().parse_args( + ["emit-spans", "--service", "nemo-test", "--span", "root,1,2"] + ) + + assert cli._run_emit_spans(args) == 0 + assert selected == [expected_exporter] + + +@pytest.mark.parametrize( + ("span_exporter", "expected_message"), + [ + (_FailingSpanExporter(), "error: failed to export spans"), + (_RaisingSpanExporter(), "error: failed to export spans"), + ], +) +def test_main_emit_spans_returns_error_when_export_fails(span_exporter, expected_message): + stderr = io.StringIO() + args = cli._build_parser().parse_args( + ["emit-spans", "--service", "nemo-test", "--span", "root,1,2"] + ) + + assert cli._run_emit_spans(args, span_exporter=span_exporter, stderr=stderr) == 2 + assert expected_message in stderr.getvalue() + + +def test_main_emit_spans_returns_error_when_flush_times_out(monkeypatch): + stderr = io.StringIO() + real_builder = cli._build_span_emitter_provider + + def build_provider(*args, **kwargs): + provider, tracked_exporter = real_builder(*args, **kwargs) + monkeypatch.setattr(provider, "force_flush", lambda: False) + return provider, tracked_exporter + + monkeypatch.setattr(cli, "_build_span_emitter_provider", build_provider) + args = cli._build_parser().parse_args( + ["emit-spans", "--service", "nemo-test", "--span", "root,1,2"] + ) + + assert ( + cli._run_emit_spans( + args, + span_exporter=InMemorySpanExporter(), + stderr=stderr, + ) + == 2 + ) + assert "error: timed out while flushing spans" in stderr.getvalue() + + +@pytest.mark.parametrize( + ("invalid_spans", "expected_message"), + [ + (["orphan,1,2,missing"], "error: invalid span 'orphan': missing parent 'missing'"), + ( + ["cycle-a,1,2,cycle-b", "cycle-b,1,2,cycle-a"], + "error: invalid span graph: parent cycle", + ), + (["root,2,3"], "error: invalid span 'root': duplicate span name"), + (["malformed,,2"], "error: skipping 'malformed': missing timestamp"), + (["backwards,2,1"], "error: skipping 'backwards': end timestamp precedes start timestamp"), + (["not-finite,nan,2"], "error: skipping 'not-finite': timestamp must be finite"), + ], +) +def test_main_emit_spans_rejects_any_invalid_input_without_partial_emission( + monkeypatch, invalid_spans, expected_message +): + stderr = io.StringIO() + exporter = InMemorySpanExporter() + + def fail_provider_setup(*args, **kwargs): + raise AssertionError("provider setup should not run for invalid span input") + + monkeypatch.setattr(cli, "_build_span_emitter_provider", fail_provider_setup) + + argv = ["emit-spans", "--service", "nemo-test", "--span", "root,1,2"] + for span in invalid_spans: + argv.extend(["--span", span]) + args = cli._build_parser().parse_args(argv) + + assert cli._run_emit_spans(args, span_exporter=exporter, stderr=stderr) == 1 + assert exporter.get_finished_spans() == [] + messages = stderr.getvalue() + assert expected_message in messages + assert "nothing emitted: invalid span input" in messages From 56c670ea2bf074e5add0971a18424db55a9dc1c4 Mon Sep 17 00:00:00 2001 From: Russell Hewett Date: Sun, 30 Aug 2026 12:25:17 -0700 Subject: [PATCH 3/3] feat(resources)!: emit v0.1 participant attrs BREAKING CHANGE: DL_RANK, DL_WORLD_SIZE, and DL_LOCAL_RANK semconv constants were removed. Use NV_DL_RANK, NV_DL_WORLD_SIZE, and NV_DL_LOCAL_RANK. Signed-off-by: Russell Hewett --- docs/design/semconv.mdx | 10 +++++----- docs/user-guide/resources.mdx | 8 ++++---- src/nemo/lens/providers.py | 6 +++--- src/nemo/lens/semconv.py | 12 ++++++++---- tests/test_providers.py | 23 ++++++++++++++++++++++- 5 files changed, 42 insertions(+), 17 deletions(-) diff --git a/docs/design/semconv.mdx b/docs/design/semconv.mdx index c0daa48..8dc4092 100644 --- a/docs/design/semconv.mdx +++ b/docs/design/semconv.mdx @@ -13,10 +13,10 @@ position: 4 Using constants instead of raw strings provides three key benefits: 1. **Grep-ability**: renaming an attribute across the codebase means changing one constant, not every call site. -2. **Type safety** (weak but real): `DL_RANK` is an exported name; typos become `ImportError`s. `"dl.rank"` typos become silent data loss. +2. **Type safety** (weak but real): `NV_DL_RANK` is an exported name; typos become `ImportError`s. `"nv.dl.rank"` typos become silent data loss. 3. **Central registry**: one file lists every attribute NeMo Lens might emit. Easy to review, easy to document. -Callers who want the string can use `DL_RANK` directly; Python strings-as-constants have no boxing cost. +Callers who want the string can use `NV_DL_RANK` directly; Python strings-as-constants have no boxing cost. ## Version Tracking @@ -68,9 +68,9 @@ Shared across Megatron-LM, NeMo RL, NeMo Gym. Anything a distributed training jo ``` nv.dl.job.uuid — stable submitted-job identifier nv.dl.run.uuid — stable run-attempt identifier -dl.rank — global rank -dl.world_size — total ranks -dl.local_rank — rank on this node +nv.dl.rank — global rank resource attribute +nv.dl.world_size — total ranks resource attribute +nv.dl.local_rank — rank on this node resource attribute dl.tensor_parallel.{rank,size} dl.pipeline_parallel.{rank,size} dl.data_parallel.{rank,size} diff --git a/docs/user-guide/resources.mdx b/docs/user-guide/resources.mdx index abb3cdb..44f8bd7 100644 --- a/docs/user-guide/resources.mdx +++ b/docs/user-guide/resources.mdx @@ -14,8 +14,8 @@ Every exporter-rank process emits these attributes (set in `providers.py:build_p | `service.name` | `config.service_name` (populated from `OTEL_SERVICE_NAME` through `NemoLensConfig.from_env()`, default `"nemo"`) | `"megatron-lm"` | | `service.version` | `nemo.lens.__version__` | `"0.1.0"`, `"0.1.0.post3+gabc1234"` | | `service.instance.id` | `"{run_id}-rank{rank}"` | `"abc123-rank0"` | -| `dl.rank` | `rank` argument | `0` | -| `dl.world_size` | `world_size` argument | `8` | +| `nv.dl.rank` | `rank` argument | `0` | +| `nv.dl.world_size` | `world_size` argument | `8` | | `nemo.run.id` | `config.run_id` (auto-generated if empty) | `"abc123"` | | `nemo.user.id` | `config.user` (if set) | `"my-team"` | | `deployment.environment` | `DEPLOYMENT_ENV` or `ENVIRONMENT` env var | `"production"` | @@ -191,7 +191,7 @@ as consumer defaults for those keys. ### Filter by Global Rank -In Jaeger, use: `dl.rank=0` +In Jaeger, use: `nv.dl.rank=0` ### Compare Distinct Runs @@ -206,7 +206,7 @@ Because these are resource attributes instead of span attributes, they apply to ## Attribute Conventions - **Use standard names.** Apply standard OTel attribute names where they exist, such as `service.*`, `k8s.*`, and `host.*`. -- **Use distributed learning prefix.** Apply the `dl.*` (distributed learning) prefix for training-specific attributes that are shared across consumers. +- **Use distributed learning prefixes.** Apply the `nv.dl.*` prefix for participant resource attributes and the `dl.*` prefix for other training-specific attributes that are shared across consumers. - **Use project-specific prefixes.** Apply the `<project>.*` prefix for project-specific attributes, such as `megatron.*`, `rl.*`, and `gym.*`. See [semconv](../design/semconv.mdx) for the full attribute namespace conventions. diff --git a/src/nemo/lens/providers.py b/src/nemo/lens/providers.py index c9743f0..1c1a11f 100644 --- a/src/nemo/lens/providers.py +++ b/src/nemo/lens/providers.py @@ -28,7 +28,7 @@ import threading from typing import TYPE_CHECKING -from nemo.lens.semconv import NEMO_SPAN_TRUNCATED +from nemo.lens.semconv import NEMO_SPAN_TRUNCATED, NV_DL_RANK, NV_DL_WORLD_SIZE if TYPE_CHECKING: from nemo.lens.config import NemoLensConfig @@ -257,8 +257,8 @@ def build_providers( attrs = { "service.name": config.service_name, "service.version": __version__, - "dl.rank": rank, - "dl.world_size": world_size, + NV_DL_RANK: rank, + NV_DL_WORLD_SIZE: world_size, } # Run identification — shared across all ranks in a job. if config.run_id: diff --git a/src/nemo/lens/semconv.py b/src/nemo/lens/semconv.py index 9dc3af8..2a89965 100644 --- a/src/nemo/lens/semconv.py +++ b/src/nemo/lens/semconv.py @@ -49,12 +49,16 @@ # wandb.* — NeMo custom (stable within NeMo ecosystem) # ------------------------------------------------------------------ # -# Distributed learning (dl.*) +# Distributed learning participant resources (nv.dl.*) # ------------------------------------------------------------------ # -DL_RANK = "dl.rank" -DL_WORLD_SIZE = "dl.world_size" -DL_LOCAL_RANK = "dl.local_rank" +NV_DL_RANK = "nv.dl.rank" +NV_DL_WORLD_SIZE = "nv.dl.world_size" +NV_DL_LOCAL_RANK = "nv.dl.local_rank" + +# ------------------------------------------------------------------ # +# Distributed learning attributes (dl.*) +# ------------------------------------------------------------------ # DL_DATA_PARALLEL_RANK = "dl.data_parallel.rank" DL_DATA_PARALLEL_SIZE = "dl.data_parallel.size" DL_TENSOR_PARALLEL_RANK = "dl.tensor_parallel.rank" diff --git a/tests/test_providers.py b/tests/test_providers.py index 293009e..b0381f3 100644 --- a/tests/test_providers.py +++ b/tests/test_providers.py @@ -35,7 +35,7 @@ build_noop_providers, build_providers, ) -from nemo.lens.semconv import NEMO_SPAN_TRUNCATED, SLURM_JOB_ID +from nemo.lens.semconv import NEMO_SPAN_TRUNCATED, NV_DL_RANK, NV_DL_WORLD_SIZE, SLURM_JOB_ID class TestBuildNoopProviders: @@ -95,6 +95,27 @@ def test_launch_resource_attributes_override_caller_defaults(self, monkeypatch): spans = custom_exporter.get_finished_spans() assert spans[0].resource.attributes[SLURM_JOB_ID] == "launch" + def test_rank_resource_attributes_use_v01_names(self, monkeypatch): + from opentelemetry.sdk.trace.export.in_memory_span_exporter import InMemorySpanExporter + + monkeypatch.delenv("OTEL_RESOURCE_ATTRIBUTES", raising=False) + + exporter = InMemorySpanExporter() + cfg = NemoLensConfig(enabled=True, exporter="console") + build_providers(cfg, rank=3, world_size=16, span_exporter=exporter) + + tracer = trace.get_tracer("test") + with tracer.start_as_current_span("ranked"): + pass + trace.get_tracer_provider().force_flush() + + (span,) = exporter.get_finished_spans() + resource_attrs = span.resource.attributes + assert resource_attrs[NV_DL_RANK] == 3 + assert resource_attrs[NV_DL_WORLD_SIZE] == 16 + assert "dl.rank" not in resource_attrs + assert "dl.world_size" not in resource_attrs + def test_traces_disabled(self): cfg = NemoLensConfig(enabled=True, exporter="console", traces_enabled=False) build_providers(cfg, rank=0, world_size=1)