feat(orchestration): route Gym's Lens telemetry through the job collector

fern/versions/latest/pages/observability/configuration.mdx

@@ -135,6 +135,10 @@ Two consequences are worth knowing:
 
 - A variable you export in your shell reaches every server, because the orchestrator
   copies its own environment into each child.
+- A job submitted with `gym eval submit` needs none of this by hand: the submit sets
+  `NEMO_GYM_OTEL_ENABLED`, the run id, the span groups, the logs switch and the exporter endpoint
+  for the driver from its `observability:` block, pointing at the collector that runs beside the
+  job. See [Collector for Submitted Jobs](/main/observability/eval-submit-collector).
 - The orchestrator generates one `run_id` and shares it with the fleet, so every process
   in a run reports the same `nemo.run.id` resource attribute.
 

fern/versions/latest/pages/observability/eval-submit-collector.mdx

@@ -20,6 +20,9 @@ otel:
   binary: /shared/otelcol-contrib     # static collector binary reachable on the compute nodes
   container: null                     # optional image instead of the host binary
   component: gym-vllm                 # display name for the scraped engine metrics
+  gym_telemetry: true                 # Gym's own nemo-lens telemetry; needs driver.gym_install (see below)
+  gym_span_groups: default,verify     # Gym span groups switched on in the job (add `sandbox` for sandboxed benchmarks)
+  gym_logs: true                      # ship Gym's Python logging as OTel logs too
   gpu_metrics_port: 9400              # dcgm-exporter on the node; null to skip
   node_metrics_port: 9100             # node_exporter on the node; null to skip
   scrape_interval_seconds: 15
@@ -30,7 +33,7 @@ The block is named `otel` because it configures the OpenTelemetry Collector only
 evidence (`observability_enabled`) and the in-process SDK (`telemetry.enabled`, `NEMO_GYM_OTEL_*`)
 are separate settings. A deployment usually puts `endpoint`, `service_name`, `token_env` and `binary` in a shared cluster
 fragment; user configs rarely need to mention the block. Set `enabled: false` to opt a run out. A job
-without a model service gets no collector.
+without a model service still gets a collector, because Gym's own servers produce telemetry.
 
 The submit fails before staging anything when `enabled` is true and `endpoint` or `service_name` is
 unset, or when the token variable is missing from the submitting shell. The error names the key to set.
@@ -42,6 +45,16 @@ check before the services start. By default it runs as a host process from `bina
 collector image is distroless and cannot start under enroot, so use `container` only with an image
 that has a shell.
 
+Gym's own telemetry needs `driver.gym_install`: the `telemetry` extra is installed with that checkout, and a
+container's preinstalled Gym is left alone. The submit fails when `gym_telemetry` is on without it; set
+`gym_telemetry: false` to keep the collector for engine and node metrics only.
+
+The driver is installed with the `telemetry` extra and started with the environment that switches
+Gym's Lens instrumentation on and points it at the collector (`NEMO_GYM_OTEL_ENABLED`,
+`NEMO_GYM_OTEL_RUN_ID` set to the gym job id, `NEMO_GYM_OTEL_SPAN_GROUPS`, `NEMO_GYM_OTEL_LOGS_ENABLED`,
+`OTEL_EXPORTER_OTLP_ENDPOINT`). Every Gym server the driver starts inherits it, so one trace spans the
+agent, model and resources servers of a rollout. An explicit value in `driver.env` wins.
+
 After the driver exits, the script waits one scrape interval so the final counters are collected,
 signals the collector process (not its `srun`), waits for a clean shutdown, and exits with the
 driver's return code.
@@ -52,11 +65,14 @@ driver's return code.
 | --- | --- | --- |
 | Model services (`type: vllm`, ...) | Prometheus scrape of `localhost:<port>/metrics` | `vllm:<name>` is renamed to `vllm_<name>` after parsing so histogram and counter types survive |
 | `dcgm-exporter`, `node_exporter` | Prometheus scrape on the node when the ports are set | per-GPU activity, memory and power; CPU, memory, network |
-| Anything in the job | OTLP traces and logs on `localhost:4317` (gRPC) and `:4318` (HTTP) | the on-ramp for Gym's own Lens telemetry |
+| Gym's servers (nemo-lens) | OTLP on `localhost:4317` (gRPC) and `:4318` (HTTP) | `gym.rollout.duration_ms`, `gym.verify.*`, FastAPI request metrics, one trace per rollout, log records with trace ids; see [Metrics](/main/observability/metrics) and [Span Groups](/main/observability/span-groups) |
+| Spans, as metrics | `span_metrics` connector | every span becomes a duration histogram and a call counter (`traces_span_metrics_*`) keyed by span name, producer and, for sandboxes, provider, so sandbox start/exec latency and error rates need no metric of their own |
 
 Every batch carries resource attributes the dashboards filter on: `user`, `run_id` (the gym job id),
 `slurm_job_id`, `benchmark`, `cluster` and `model`, plus `service.name` set to the routing identity
-and `service.name.override` to the producer's display name.
+and `service.name.override` to the producer's display name: each Gym server keeps its own name
+(`nemo-gym/<server>`), a scraped model service carries `<component>/<service>` (`gym-vllm/policy` with the
+defaults), and the node scrapes carry `dcgm` and `node`.
 
 Everything exported is also written to `<job dir>/otel/{metrics,traces,logs}.jsonl`, so a run can be
 inspected without backend access. The rendered collector configuration is at

nemo_gym/orchestration/api.py

@@ -312,6 +312,12 @@ class OtelConfig(_StrictModel):
     service_name: str | None = None
     # Display identity of the scraped metrics in the backend (`service.name.override`).
     component: str = "gym-vllm"
+    # Gym's own nemo-lens telemetry; needs `driver.gym_install`, which installs the `telemetry` extra.
+    gym_telemetry: bool = True
+    # Gym span groups to switch on: a preset or comma-separated names (`default`, `verify`, `sandbox`, ...).
+    gym_span_groups: str = "default,verify"
+    # Ship Gym's Python logging as OTel logs too (trace-correlated), through the same collector.
+    gym_logs: bool = True
     # Node-level exporters that clusters commonly run as system services on every compute node;
     # scraped on localhost when set, skipped when null. DCGM gives per-GPU activity/memory/power,
     # node_exporter gives CPU/memory/network/disk. A closed port only logs scrape errors.

nemo_gym/orchestration/executors/otel.py

@@ -41,6 +41,28 @@
 COLLECTOR_HEALTH_PORT = 13133
 OTLP_GRPC_PORT = 4317
 OTLP_HTTP_PORT = 4318
+GYM_TELEMETRY_EXTRA = "telemetry"
+
+
+def driver_telemetry_env(gym_job_id: str, span_groups: str, *, logs: bool = True) -> dict[str, str]:
+    """Environment that switches on Gym's Lens instrumentation and points it at the collector.
+
+    Gym reads these in every server process (`NEMO_GYM_OTEL_*` are Gym's own, the `OTEL_*` ones
+    are the SDK's); an explicit value in `driver.env` wins over these.
+    """
+    return {
+        "NEMO_GYM_OTEL_ENABLED": "1",
+        "NEMO_GYM_OTEL_RUN_ID": gym_job_id,
+        "NEMO_GYM_OTEL_SPAN_GROUPS": span_groups,
+        "NEMO_GYM_OTEL_LOGS_ENABLED": "1" if logs else "0",
+        "OTEL_EXPORTER_OTLP_ENDPOINT": f"http://localhost:{OTLP_HTTP_PORT}",
+        "OTEL_EXPORTER_OTLP_PROTOCOL": "http/protobuf",
+        # nemo-lens builds its log exporter over gRPC regardless of the protocol setting, so logs
+        # get the collector's gRPC port explicitly (the SDK's per-signal endpoint takes precedence).
+        "OTEL_EXPORTER_OTLP_LOGS_ENDPOINT": f"http://localhost:{OTLP_GRPC_PORT}",
+    }
+
+
 # Seconds the collector keeps running after the driver exits, so one more scrape sees the final
 # counters before it is asked to flush and stop.
 FINAL_SCRAPE_GRACE_SECONDS = 20
@@ -55,8 +77,23 @@ def scrape_targets(config: SubmitConfig) -> dict[str, int]:
 
 
 def otel_active(config: SubmitConfig) -> bool:
-    """Whether a collector step is added to this job: enabled, and there is something to scrape."""
-    return config.otel.enabled and bool(scrape_targets(config))
+    """Enabled is enough: Gym's own servers push telemetry even when there is no local model to scrape."""
+    return config.otel.enabled
+
+
+def gym_telemetry_active(config: SubmitConfig) -> bool:
+    """Whether the driver is instrumented with nemo-lens and pointed at the collector."""
+    return config.otel.enabled and config.otel.gym_telemetry
+
+
+def validate_gym_telemetry(config: SubmitConfig) -> None:
+    if gym_telemetry_active(config) and config.driver.gym_install is None:
+        raise ValueError(
+            "otel.gym_telemetry is on but driver.gym_install is not set, so the driver cannot install "
+            "Gym's telemetry extra and no Gym metrics or traces would be exported. Set driver.gym_install "
+            "(repo, ref), or set `otel.gym_telemetry: false` to run the collector with engine and node "
+            "metrics only."
+        )
 
 
 def validate_destination(config: SubmitConfig) -> None:
@@ -107,9 +144,10 @@ def render_collector_config(config: SubmitConfig, benchmark_name: str, remote_be
     token = f"${{env:{obs.token_env}}}"
     interval = f"{obs.scrape_interval_seconds}s"
 
+    # The job name is what scraped data carries as `service.name`, and so becomes its display identity.
     scrape_configs = [
         {
-            "job_name": f"vllm-{name}",
+            "job_name": f"{obs.component}/{name}",
             "scrape_interval": interval,
             "static_configs": [{"targets": [f"localhost:{port}"], "labels": {"gym_service": name}}],
         }
@@ -126,6 +164,15 @@ def render_collector_config(config: SubmitConfig, benchmark_name: str, remote_be
                     "static_configs": [{"targets": [f"localhost:{port}"]}],
                 }
             )
+    # Each producer's own `service.name` survives as the display identity before `resource` overwrites it.
+    keep_display_name = (
+        'set(resource.attributes["service.name.override"], resource.attributes["service.name"]) '
+        'where resource.attributes["service.name.override"] == nil and resource.attributes["service.name"] != nil'
+    )
+    identity = {
+        f"{signal}_statements": [{"context": "resource", "statements": [keep_display_name]}]
+        for signal in ("metric", "trace", "log")
+    }
     # vLLM names its metrics `vllm:<name>`; the shared dashboards, and Prometheus convention, use
     # `vllm_<name>`, and the backend keeps whatever name arrives. Renamed after parsing so counters
     # and histograms keep their types (Prometheus relabelling would make them untyped). `$$` escapes
@@ -141,8 +188,6 @@ def render_collector_config(config: SubmitConfig, benchmark_name: str, remote_be
 
     attributes = [
         ("service.name", obs.service_name, "upsert"),
-        # `insert` so a producer that already names its own component keeps it.
-        ("service.name.override", obs.component, "insert"),
         ("Authorization", token, "upsert"),
         ("user", getpass.getuser(), "upsert"),
         ("run_id", remote_bench_dir.parent.name, "upsert"),
@@ -158,10 +203,20 @@ def render_collector_config(config: SubmitConfig, benchmark_name: str, remote_be
     # dashboards match it as a label string.
     resource_actions.append({"key": "slurm_job_id", "action": "convert", "converted_type": "string"})
 
+    # Operations that exist only as spans (sandbox start/exec, model calls) still get latency and count series.
+    span_metrics = {
+        "histogram": {
+            "explicit": {"buckets": ["250ms", "1s", "2s", "5s", "10s", "30s", "60s", "120s", "300s", "600s", "1800s"]}
+        },
+        "dimensions": [{"name": "service.name.override"}, {"name": "nemo.gym.sandbox.provider"}],
+        "metrics_flush_interval": f"{obs.scrape_interval_seconds}s",
+    }
+
     doc = {
         "extensions": {"health_check": {"endpoint": f"0.0.0.0:{COLLECTOR_HEALTH_PORT}"}},
+        "connectors": {"span_metrics": span_metrics},
         "receivers": {
-            "prometheus": {"config": {"scrape_configs": scrape_configs}},
+            **({"prometheus": {"config": {"scrape_configs": scrape_configs}}} if scrape_configs else {}),
             "otlp": {
                 "protocols": {
                     "grpc": {"endpoint": f"0.0.0.0:{OTLP_GRPC_PORT}"},
@@ -172,6 +227,7 @@ def render_collector_config(config: SubmitConfig, benchmark_name: str, remote_be
         "processors": {
             "batch": {},
             "resource": {"attributes": resource_actions},
+            "transform/identity": identity,
             "transform/metric_names": rename_colon_metrics,
         },
         "exporters": {
@@ -187,18 +243,18 @@ def render_collector_config(config: SubmitConfig, benchmark_name: str, remote_be
             "telemetry": {"logs": {"level": "debug"}},
             "pipelines": {
                 "metrics": {
-                    "receivers": ["prometheus"],
-                    "processors": ["transform/metric_names", "resource", "batch"],
+                    "receivers": (["prometheus"] if scrape_configs else []) + ["otlp", "span_metrics"],
+                    "processors": ["transform/metric_names", "transform/identity", "resource", "batch"],
                     "exporters": ["otlp_http/managed", "file/metrics"],
                 },
                 "traces": {
                     "receivers": ["otlp"],
-                    "processors": ["resource", "batch"],
-                    "exporters": ["otlp_http/managed", "file/traces"],
+                    "processors": ["transform/identity", "resource", "batch"],
+                    "exporters": ["otlp_http/managed", "file/traces", "span_metrics"],
                 },
                 "logs": {
                     "receivers": ["otlp"],
-                    "processors": ["resource", "batch"],
+                    "processors": ["transform/identity", "resource", "batch"],
                     "exporters": ["otlp_http/managed", "file/logs"],
                 },
             },

nemo_gym/orchestration/executors/script_templates.py

@@ -178,6 +178,8 @@ def render_driver_entrypoint(
     ref: str | None,
     prepare_cmd: str | None,
     command: str | None = None,
+    *,
+    extras: tuple[str, ...] = (),
 ) -> str:
     """Render the srun entrypoint for the driver step.
 
@@ -187,8 +189,14 @@ def render_driver_entrypoint(
     `command` replaces the `gym eval run` invocation for a benchmark whose harness
     is not Gym's own runner. It is emitted verbatim, so it may be a multi-line
     script; everything before it (the install, and prepare) is unchanged.
+
+    `extras` are optional-dependency groups installed with the checkout; the server
+    venvs Gym builds copy their telemetry packages from this process, so the driver
+    is where `telemetry` has to be installed for any server to have it.
     """
     preamble: list[str] = []
+    # Double quotes: the whole preamble ends up inside a single-quoted `bash -c` block.
+    install_target = f'".[{",".join(extras)}]"' if extras else "."
 
     if repo and ref:
         # Clone to /tmp rather than the job directory: a checkout plus its .venv
@@ -208,7 +216,7 @@ def render_driver_entrypoint(
             # EXTERNALLY-MANAGED (PEP 668), so no --break-system-packages override needed either.
             "uv venv --seed .venv",
             "source .venv/bin/activate",
-            "uv pip install -e .",
+            f"uv pip install -e {install_target}",
         ]
 
     if prepare_cmd:

nemo_gym/orchestration/executors/slurm.py

@@ -31,6 +31,7 @@
     render_collector_config,
     resolve_token,
     validate_destination,
+    validate_gym_telemetry,
 )
 from nemo_gym.orchestration.executors.slurm_script import build_sbatch_script
 from nemo_gym.orchestration.jobs import (
@@ -158,6 +159,7 @@ def run(self, config: SubmitConfig, *, dry_run: bool = False) -> SubmissionRecor
         token = None
         if otel_active(config):
             validate_destination(config)
+            validate_gym_telemetry(config)
             token = resolve_token(config)
         now = utc_now()
         gym_job_id = new_gym_job_id(now)

nemo_gym/orchestration/executors/slurm_script.py

@@ -34,8 +34,11 @@
     COLLECTOR_HEALTH_PORT,
     COLLECTOR_SERVICE_NAME,
     FINAL_SCRAPE_GRACE_SECONDS,
+    GYM_TELEMETRY_EXTRA,
     SHUTDOWN_WAIT_SECONDS,
     collector_config_path,
+    driver_telemetry_env,
+    gym_telemetry_active,
     otel_active,
 )
 from nemo_gym.orchestration.executors.script_templates import (
@@ -789,6 +792,7 @@ def build_sbatch_script(
     )
 
     observed = otel_active(config)
+    instrumented = gym_telemetry_active(config)
 
     driver_node = _driver_node(config, compute) if is_multi_node else None
     service_commands = "\n\n".join(
@@ -854,6 +858,13 @@ def build_sbatch_script(
     policy_type = config.driver.policy_model_type
     extra_flags = [f"--model-type {shlex.quote(policy_type)}"] if config.driver.policy_model and policy_type else []
     driver_env = dict(config.driver.env)
+    if instrumented:
+        driver_env = {
+            **driver_telemetry_env(
+                remote_bench_dir.parent.name, config.otel.gym_span_groups, logs=config.otel.gym_logs
+            ),
+            **driver_env,
+        }
     if benchmark.command is None:
         run_args = _with_default_capture_dir(benchmark.run, remote_bench_dir)
         run_args.setdefault("require_complete", True)
@@ -870,6 +881,7 @@ def build_sbatch_script(
         ref=gi.ref if gi else None,
         prepare_cmd=prepare_cmd,
         command=benchmark.command,
+        extras=(GYM_TELEMETRY_EXTRA,) if instrumented else (),
     )
     prepare_command = ""
     driver_env_prefix = _resolve_env(driver_env) if driver_env else ""