NVIDIA-NeMo · andreatgretel · Jul 8, 2026 · Jul 7, 2026 · Jul 7, 2026 · Jul 7, 2026
@@ -399,6 +399,55 @@ except DataDesignerEarlyShutdownError:
     ...
 ```
 
+## Local OpenTelemetry Metrics
+
+`DataDesigner.create()` starts or reuses a process-wide OpenTelemetry Prometheus endpoint on `http://127.0.0.1:9464/metrics` by default. Use `RunConfig` to disable metrics for an invocation or choose another port:
+
+```python
+import data_designer.config as dd
+from data_designer.interface import DataDesigner
+
+designer = DataDesigner()
+designer.set_run_config(dd.RunConfig(otel_metrics_port=None))
+
+# Or keep metrics enabled on a different loopback port.
+designer.set_run_config(dd.RunConfig(otel_metrics_port=9465))
+```
+
+Configure a Prometheus server on the same host to scrape the endpoint:
+
+```yaml
+scrape_configs:
+  - job_name: data-designer
+    metrics_path: /metrics
+    static_configs:
+      - targets: ["127.0.0.1:9464"]
+```
+
+The endpoint exposes seven instruments. Prometheus derives throughput and request-completion rates from their counters and histogram counts.
+
+| Instrument | Type | Meaning |
+|------------|------|---------|
+| `data_designer.create.duration` | Histogram | Create-job duration in seconds, with failure-only `error.type`. |
+| `data_designer.dataset.records` | Counter | Checkpointed generated and dropped records, distinguished by `record.result`; derive record throughput from this counter. |
+| `data_designer.dataset.progress` | Gauge | Fraction of scheduled records processed by the current create job, or across all active jobs when creates overlap. The most recently finished value remains visible until the next job starts. |
+| `data_designer.scheduler.events` | Counter | Job start/completion, non-retryable task errors, worker spawn failures, deferred retries, and task cancellations, identified by `event.name` with bounded `outcome` and failure-only `error.type` where applicable. |
+| `data_designer.model.request.active` | UpDownCounter | Model requests currently in progress, identified by `gen_ai.operation.name`, `gen_ai.provider.name`, and `gen_ai.request.model`. |
+| `gen_ai.client.operation.duration` | Histogram | Completed model-request duration in seconds, identified by `gen_ai.operation.name`, `gen_ai.provider.name`, and `gen_ai.request.model`, with failure-only `error.type`. |
+| `data_designer.log.records` | Counter | Data Designer log-record counts by normalized `log.severity`. |
+
+The listener and cumulative metrics remain available until process shutdown so a scrape can collect results after a job finishes. A later enabled job can rebind an idle listener to a different configured port without resetting those metrics. Concurrent jobs that request different ports share the active listener and emit a warning instead of replacing it.
+
+Generated and dropped record counts advance when a row group is durably checkpointed, so `data_designer.dataset.progress` moves in `buffer_size`-sized steps. Use `data_designer.model.request.active` and the `gen_ai.client.operation.duration` histogram count for live request activity between checkpoints.
+
+Setting `otel_metrics_port=None` records nothing for that invocation. If an earlier enabled job already started the process listener, disabling a later job does not stop the listener or remove the earlier cumulative metrics.
+
+<Warning>
+The endpoint binds only to loopback and has no authentication. It is intended for a Prometheus server running on the same host, not for remote exposure.
+</Warning>
+
+Prometheus pull is metrics-only: `data_designer.log.records` counts safe log metadata, not raw log bodies. Existing raw logs continue to use the configured stdout and file handlers and should be collected from those outputs. The Python Prometheus exporter does not support multiprocessing, so this endpoint is unsupported for multiprocessing-based collection.
+
 ## Common Problems
 
 | Problem | Symptom | Solution |

@@ -159,6 +159,10 @@ class RunConfig(ConfigBase):
             non-TTY environments. Default is False.
         progress_interval: How often (in seconds) the async progress reporter emits a
             consolidated log block. Must be > 0. Default is 5.0.
+        otel_metrics_port: Loopback port for the pull-based OpenTelemetry metrics endpoint.
+            Set to None to disable instrumentation for this create invocation. An endpoint
+            already opened by the process may remain available for metrics from prior runs.
+            The endpoint exposes metrics, not raw log records. Default is 9464.
         preserve_dropped_columns: If True, write columns removed by drop processors to
             separate dropped-column parquet files. Set to False to omit those artifacts
             while still removing dropped columns from the final dataset. Default is True.
@@ -200,6 +204,16 @@ class RunConfig(ConfigBase):
     write_scheduler_events: bool = False
     display_tui: bool = False
     progress_interval: float = Field(default=5.0, gt=0.0)
+    otel_metrics_port: int | None = Field(
+        default=9464,
+        ge=1,
+        le=65535,
+        description=(
+            "Loopback port for the pull-based OpenTelemetry metrics endpoint. "
+            "None disables instrumentation for this create invocation; an existing process endpoint may remain "
+            "available for prior metrics. Raw log records are not exposed."
+        ),
+    )
     preserve_dropped_columns: bool = Field(
         default=True,
         description=(

@@ -40,6 +40,36 @@ def test_run_config_accepts_scheduler_event_writes() -> None:
     assert RunConfig(write_scheduler_events=True).write_scheduler_events is True
 
 
+def test_run_config_defaults_otel_metrics_port_to_9464() -> None:
+    assert RunConfig().otel_metrics_port == 9464
+
+
+def test_run_config_accepts_custom_otel_metrics_port() -> None:
+    assert RunConfig(otel_metrics_port=4318).otel_metrics_port == 4318
+
+
+def test_run_config_accepts_disabled_otel_metrics() -> None:
+    assert RunConfig(otel_metrics_port=None).otel_metrics_port is None
+
+
+@pytest.mark.parametrize("otel_metrics_port", [1, 65535])
+def test_run_config_accepts_otel_metrics_port_bounds(otel_metrics_port: int) -> None:
+    assert RunConfig(otel_metrics_port=otel_metrics_port).otel_metrics_port == otel_metrics_port
+
+
+@pytest.mark.parametrize("otel_metrics_port", [0, 65536])
+def test_run_config_rejects_otel_metrics_port_outside_bounds(otel_metrics_port: int) -> None:
+    with pytest.raises(ValidationError, match="otel_metrics_port"):
+        RunConfig(otel_metrics_port=otel_metrics_port)
+
+
+def test_run_config_preserves_otel_metrics_port_when_serialized() -> None:
+    serialized = RunConfig(otel_metrics_port=4318).model_dump()
+
+    assert serialized["otel_metrics_port"] == 4318
+    assert RunConfig.model_validate(serialized).otel_metrics_port == 4318
+
+
 def test_run_config_progress_bar_shim_translates_to_display_tui() -> None:
     with pytest.warns(DeprecationWarning, match="RunConfig.progress_bar.*RunConfig.display_tui") as caught:
         run_config = RunConfig(progress_bar=False)

@@ -77,8 +77,10 @@
 from data_designer.engine.observability import (
     RuntimeCorrelation,
     SchedulerAdmissionEvent,
+    SchedulerAdmissionEventKind,
     SchedulerAdmissionEventSink,
     runtime_correlation_provider,
+    scheduler_event_sink_accepts,
 )
 from data_designer.engine.processing.ginja.exceptions import UserTemplateError
 from data_designer.engine.progress.reporter import (
@@ -235,7 +237,14 @@ def __init__(
         self._in_flight: set[Task] = set()
         self._worker_tasks: set[asyncio.Task] = set()
         self._wake_event = asyncio.Event()
-        self._run_id = run_id or f"run-{uuid.uuid4().hex}"
+        ambient_correlation = runtime_correlation_provider.current()
+        self._run_id = (
+            run_id
+            if run_id is not None
+            else ambient_correlation.run_id
+            if ambient_correlation is not None
+            else f"run-{uuid.uuid4().hex}"
+        )
         self._scheduler_event_sink = scheduler_event_sink
         self._scheduler_event_sequence = 0
         if salvage_max_rounds < 1:
@@ -448,7 +457,7 @@ def _spawn_worker(self, coro: Coroutine[Any, Any, None]) -> asyncio.Task:
 
     def _emit_scheduler_event(
         self,
-        event_kind: str,
+        event_kind: SchedulerAdmissionEventKind,
         *,
         task: Task | None = None,
         lease: TaskAdmissionLease | None = None,
@@ -457,10 +466,18 @@ def _emit_scheduler_event(
         reason_or_result: str | None = None,
         diagnostics: dict[str, object] | None = None,
     ) -> None:
-        if self._scheduler_event_sink is None:
+        if not scheduler_event_sink_accepts(self._scheduler_event_sink, event_kind):
             return
         self._scheduler_event_sequence += 1
-        correlation = None
+        correlation = RuntimeCorrelation(
+            run_id=self._run_id,
+            row_group=None,
+            task_column=None,
+            task_type=None,
+            scheduling_group_kind=None,
+            scheduling_group_identity_hash=None,
+            task_execution_id=None,
+        )
         event_diagnostics = dict(diagnostics or {})
         if task is not None:
             schedulable = lease.item if lease is not None else self._schedulable_task(task)
@@ -480,7 +497,7 @@ def _emit_scheduler_event(
         try:
             self._scheduler_event_sink.emit_scheduler_event(
                 SchedulerAdmissionEvent.capture(
-                    event_kind,  # type: ignore[arg-type]
+                    event_kind,
                     sequence=self._scheduler_event_sequence,
                     correlation=correlation,
                     task_id=stable_task_id(task) if task is not None else None,
@@ -530,7 +547,7 @@ def _record_observed_task_state(self) -> None:
             )
 
     def _emit_scheduler_health_snapshot(self, reason: str) -> None:
-        if self._scheduler_event_sink is None:
+        if not scheduler_event_sink_accepts(self._scheduler_event_sink, "scheduler_health_snapshot"):
             return
         self._emit_scheduler_event(
             "scheduler_health_snapshot",
@@ -1244,9 +1261,11 @@ async def run(self) -> None:
 
         num_rgs = len(self._row_groups)
         self._run_loop = asyncio.get_running_loop()
+        outcome = "success"
+        error_type: str | None = None
 
-        with self._progress_bar or contextlib.nullcontext():
-            try:
+        try:
+            with self._progress_bar or contextlib.nullcontext():
                 if self._reporter:
                     self._reporter.log_start(num_row_groups=num_rgs, scheduled_records=self._scheduled_records)
 
@@ -1284,20 +1303,38 @@ async def run(self) -> None:
                     self._reporter.log_final()
 
                 self._emit_scheduler_health_snapshot("completed")
-                self._emit_scheduler_event(
-                    "scheduler_job_completed", diagnostics=self._scheduler_health_diagnostics(reason="completed")
-                )
 
                 if self._rg_states:
                     incomplete = list(self._rg_states)
                     logger.error(
                         f"Scheduler exited with {len(self._rg_states)} unfinished row group(s): {incomplete}. "
                         "These row groups were not checkpointed."
                     )
-            finally:
-                if self._reporter:
-                    self._reporter.close()
-                self._run_loop = None
+        except asyncio.CancelledError:
+            outcome = "cancelled"
+            raise
+        except BaseException as exc:
+            outcome = "error"
+            error_type = type(exc).__name__
+            raise
+        finally:
+            self._run_loop = None
+            if outcome == "success" and self._early_shutdown:
+                outcome = "early_shutdown"
+            terminal_diagnostics: dict[str, object] = {"outcome": outcome}
+            if outcome in {"success", "early_shutdown"} and scheduler_event_sink_accepts(
+                self._scheduler_event_sink, "scheduler_job_completed"
+            ):
+                terminal_diagnostics = self._scheduler_health_diagnostics(reason="completed") | terminal_diagnostics
+            if error_type is not None:
+                terminal_diagnostics["error_type"] = error_type
+            self._emit_scheduler_event(
+                "scheduler_job_completed",
+                reason_or_result=outcome,
+                diagnostics=terminal_diagnostics,
+            )
+            if self._reporter:
+                self._reporter.close()
 
     async def _main_dispatch_loop(
         self,

@@ -50,7 +50,11 @@
 from data_designer.engine.dataset_builders.utils.processor_runner import ProcessorRunner, ProcessorStage
 from data_designer.engine.dataset_builders.utils.row_group_buffer import RowGroupBufferManager
 from data_designer.engine.models.telemetry import InferenceEvent, NemoSourceEnum, TaskStatusEnum, TelemetryHandler
-from data_designer.engine.observability import JsonlSchedulerEventSink, SchedulerAdmissionEventSink
+from data_designer.engine.observability import (
+    JsonlSchedulerEventSink,
+    SchedulerAdmissionEventSink,
+    fanout_scheduler_event_sinks,
+)
 from data_designer.engine.processing.processors.base import Processor
 from data_designer.engine.processing.processors.drop_columns import DropColumnsProcessor
 from data_designer.engine.readiness import run_readiness_check
@@ -984,7 +988,10 @@ def on_before_checkpoint(rg_id: int, rg_size: int) -> None:
             initial_completed_records=initial_actual_num_records,
             progress_interval=self._resource_provider.run_config.progress_interval,
             display_tui=self._resource_provider.run_config.display_tui,
-            scheduler_event_sink=scheduler_event_sink,
+            scheduler_event_sink=fanout_scheduler_event_sinks(
+                scheduler_event_sink,
+                self._resource_provider.scheduler_event_sink,
+            ),
             request_pressure_provider=self._resource_provider.model_registry.request_admission,
             request_pressure_advisory=True,
         )

@@ -54,6 +54,7 @@ def create_model_client(
             setup, which happens before any generation task invokes the client.
             Direct callers of this factory must ensure registration happens
             before use.
+        request_event_sink: Optional direct sink for model-request events.
 
     Returns:
         A ``ModelClient`` instance routed by provider type.