feat: add OpenTelemetry metrics for create jobs

[eric-tramel]

📋 Summary

What. Adds default, pull-based OpenTelemetry metrics for DataDesigner.create() on a loopback Prometheus endpoint controlled by RunConfig.

Why. Gives operators current create progress, throughput, scheduler outcomes, model activity, request completion/latency, errors, and safe log-volume signals without making task code depend on OpenTelemetry or adding parallel instrumentation paths.

The interface layer owns a private OTel meter provider, registry, and HTTP listener. Existing generic engine scheduler and request-event seams feed the exporter, while a metric-only logging handler tees severity counts without exporting log bodies. Telemetry failures are contained and cannot change create results.

🔗 Related Issue

None.

🔄 Changes

• Adds RunConfig.otel_metrics_port, defaulting to loopback port 9464; None disables telemetry for an invocation.
• Exposes seven instruments at /metrics:
  • data_designer.create.duration
  • data_designer.dataset.records
  • data_designer.dataset.progress
  • data_designer.scheduler.events
  • data_designer.model.request.active
  • gen_ai.client.operation.duration
  • data_designer.log.records
• Reports lifecycle-aware current progress and active model requests, including standard gen_ai.operation.name, gen_ai.provider.name, and gen_ai.request.model attributes.
• Extends the existing generic scheduler/request event seams with cheap event-interest filtering, terminal scheduler outcomes, request timing, and production sink wiring.
• Composes the per-run JSONL scheduler sink from current main with the OTel sink through an exception-isolated fan-out, so both observers receive events when enabled.
• Guarantees sync/async request-lease cleanup if started-event delivery is interrupted, and applies request interest gates before expensive attempt snapshots are built.
• Reports early shutdown as a distinct terminal outcome while preserving the scheduler completion diagnostics contract.
• Keeps OTel SDK/exporter ownership in the interface package and preserves config → engine dependency boundaries and lazy imports.
• Serializes run teardown with event admission and state cleanup so delayed events or overlapping creates cannot publish stale gauges.
• Retires an old listener after releasing the runtime lock so a concurrent scrape cannot deadlock an idle port rebind.
• Captures runtime state and reconciliation failures under the runtime lock, then emits lifecycle and warning logs only after releasing it to avoid lock-order inversion with logging handlers.
• Preserves exact short provider/model identities and bounds long metric attributes with stable hash-suffixed truncation, avoiding collisions while keeping the 128-character limit; excludes prompts, responses, request/run/task identifiers, exception text, environment resource attributes, and log bodies.
• Documents scrape configuration, metric semantics, checkpoint-granularity progress, listener lifecycle, security posture, and raw-log limitations.

📖 Usage

The default create path starts http://127.0.0.1:9464/metrics:

import data_designer.config as dd
from data_designer.interface import DataDesigner

designer = DataDesigner()
designer.create(config_builder)

Choose another loopback port or disable telemetry:

designer.set_run_config(dd.RunConfig(otel_metrics_port=9465))
# designer.set_run_config(dd.RunConfig(otel_metrics_port=None))

Prometheus can scrape 127.0.0.1:9464 at /metrics. The listener is process-wide and remains available after a create job so final cumulative metrics can be collected. Generated-record totals and progress advance at durable row-group checkpoints; the active-request gauge and request-duration count provide live activity between checkpoints.

📊 External dashboard validation

A local external Prometheus/Grafana stack scraped the listener every 2 seconds and rendered current listener health, active jobs, durable progress, generated records, active requests by exact model, attempts/second, p95 latency, scheduler totals, failures, and log volume.

The end-to-end demonstration generated 512 records through 1,536 logical model tasks using the nvidia-internal provider and three non-8B models, each configured with max_tokens=1024:

• nvidia/openai/gpt-oss-20b
• nvidia/nvidia/nemotron-nano-31b-v3
• nvidia/openai/gpt-oss-120b

All 1,536 logical tasks completed successfully. The external collector observed two transient internal_server attempts on the 120B model, two matching task.retry_deferred scheduler events, and final job success. Dashboard queries use raw counters for event/failure totals, short rate windows aligned with the 2-second scrape interval, a lifecycle-aware progress gauge, and model-separated GenAI series so the panels represent current state instead of long-window residual activity.

The original POC capture is retained below; the validated dashboard now includes the live-state and 2-second refresh refinements described above.

📝 Changelog

• Adds default local OpenTelemetry/Prometheus metrics for DataDesigner create jobs with a RunConfig port/disable option.

🧪 Testing

• make check-all — passed

• make test — 3,940 passed, 1 existing skip

• Focused scheduler, request-admission, builder, interface, and OpenTelemetry rebase suite — 410 passed

• Seven-persona delegated review of the latest deadlock/label patch — no actionable findings; one P4 simplification suggestion was rejected because it would remove the documented 128-character attribute bound

• Rebase-specific local review across correctness, API compatibility, and test coverage — no actionable findings

• Deterministic delayed-event, overlapping-teardown, unlocked-rebind, handler-lock inversion, and provider/model identity-collision regressions — passed

• Isolated package-install suites — passed during feature validation

• Import-performance and lazy-import checks — passed

• Fern validation — 0 errors, 1 existing warning

• Live nvidia-internal multi-model run — 512 records and 1,536 logical tasks completed successfully while an external Prometheus/Grafana stack observed execution

• make test passes

• Unit tests added/updated

• E2E behavior validated

👀 Review areas

• RunConfig surface
• Generic engine observability seam
• OpenTelemetry adapter and listener
• User documentation

✅ Checklist

• Follows commit message conventions
• Commits are signed off (DCO)
• Architecture docs updated (if applicable)

---

Description updated with AI

[github-actions]

Fern preview: https://nvidia-preview-pr-802.docs.buildwithfern.com/nemo/datadesigner

Fern previews include the docs-website version archive with PR changes synced into latest. Notebook tutorials are rendered without execution outputs in previews.

[github-actions]

Code Review: PR #802 — feat: add OpenTelemetry metrics for create jobs

Author: eric-tramel · Base: main ← codex/otel-create-observability
Size: +1,461 / −47 across 23 files · State: OPEN

Summary

This PR adds default, pull-based OpenTelemetry/Prometheus metrics for DataDesigner.create(). A process-owned OpenTelemetryRuntime (in the interface package) owns the meter provider, Prometheus CollectorRegistry, and a loopback HTTP listener bound to 127.0.0.1:9464 (configurable via RunConfig.otel_metrics_port, None to disable). Five cumulative instruments are exposed: create duration, dataset records, scheduler events, GenAI request duration, and log-severity counts.

The design correctly respects the package layering. OTel SDK/exporter ownership stays in data-designer (interface); the engine gains only generic event-sink seams (accepts_scheduler_event interest filtering, terminal scheduler outcomes, request timing), and config gains only a plain integer field. The OpenTelemetryRuntime is passed into the engine as an anonymous SchedulerAdmissionEventSink / RequestAdmissionEventSink implementer — the engine never imports the interface. This is the right seam.

The change is unusually well-tested: privacy (secret-sentinel scrubbing), lifecycle/rebind, concurrent creates, exact-once emission, init-failure cleanup, real HTTP scrape, and a subprocess test asserting the OTel SDK is not imported when disabled. Failure containment is a clear design goal and is consistently applied.

Findings

Correctness

• _non_negative_int rejects bool but silently drops floats (minor). packages/data-designer/src/data_designer/integrations/opentelemetry.py:1298 — max(0, value) if isinstance(value, int) else 0. surviving_rows/dropped_rows are ints today so this is fine, but note isinstance(True, int) is True, so a stray bool would count as 1/0. Non-blocking; current callers only pass ints.

• Import-direction violations in structural analysis are false positives. The graphify report flags 27 "config → engine .set()" violations. These are misattributions: the .set() calls are runtime_correlation_provider.set() (a ContextVar) and OTel histogram/counter .set()/.add(), not config importing engine. I verified run_config.py imports nothing from engine (the only config change is a plain int | None Pydantic field). No real layering violation exists. See Structural Impact section below.

• run refactor preserves the finally-emit-once contract correctly. async_scheduler.py:1082+ restructures the try/with into try/except/finally so the terminal scheduler_job_completed event fires exactly once with outcome in {success, cancelled, error} and a failure-only error_type. The previous duplicate emit (health snapshot + separate job-completed inside the body) is removed. CancelledError is re-raised without being swallowed. Test test_scheduler_emits_one_terminal_event_without_replacing_failure covers both the error and cancelled paths and asserts the original exception is re-raised (exc_info.value is raised). Good.

• model_request_executor timing refactor is sound. Both sync and async paths move to a single finally that emits model_request_completed once with outcome + duration_seconds. The nested inner try/finally around call() ensures duration_seconds is captured even when the call raises, before the outer handler classifies the outcome. outcome is initialized to "unexpected_exception" so an unexpected escape still emits a defined value. Correct.

Design / Architecture

• Ambient run-id inheritance is the linchpin — worth calling out. observe_create sets a runtime_correlation_provider ContextVar with a fresh run-<uuid>; the scheduler now inherits that ambient run_id when no explicit run_id is passed (async_scheduler.py:230), and _event_is_active / record_log gate on membership in _active_run_ids. This is how metrics get scoped to a specific create invocation across threads. The concurrency test (test_concurrent_data_designer_creates_share_one_exporter) and the log-scoping test exercise this. The mechanism is subtle but correct and well-covered.

• Repeated self._open_telemetry if self._run_config.otel_metrics_port is not None else None (3×) in data_designer.py (lines 745, 746, 754). Minor DRY nit — a small _active_event_sink() helper would remove the triplication and keep the enable/disable rule in one place. Non-blocking.

• getattr(self._resource_provider, "scheduler_event_sink", None) in dataset_builder.py:976 uses a defensive getattr even though ResourceProvider now declares the field. Since it's a typed model attribute, self._resource_provider.scheduler_event_sink would be safe and clearer. Presumably defensive for stub/mock providers in tests — acceptable, but a direct access would be more honest about the contract.

• Log handler attaches to the data_designer root logger process-wide. _initialize adds _MetricLogHandler to logging.getLogger("data_designer"). record_log correctly gates on the active-run set, so counts are scoped, but the handler stays attached for the process lifetime (removed only on shutdown/atexit). This is intentional (documented "process-wide listener remains available"), and the handler is cheap + exception-swallowing. Fine, but reviewers should be aware every data_designer.* log record now passes through this handler once metrics have been initialized once.

Security / Privacy

• Privacy posture is strong and explicitly tested. _bounded_attribute whitelists [alnum . _ -], truncates to 128 chars, and collapses to _OTHER; unknown GenAI operations and domains collapse to _OTHER; model IDs, request/attempt IDs, and raw log bodies are never exported. test_runtime_exports_bounded_metrics_without_log_bodies asserts no SECRET-* sentinel (including OTEL_RESOURCE_ATTRIBUTES injection) leaks into the scrape, and target_info is restricted to service_name/service_version. Error text is excluded — only error.type (class name) is recorded. This is exactly right for a metrics endpoint.

• Loopback-only bind + documented warning. Binds to 127.0.0.1 only, with a docs <Warning> that there is no auth and it's not for remote exposure. Reasonable for the stated same-host-Prometheus use case. Defaulting the endpoint on (port 9464) is a deliberate product choice; operators who object can set None. Worth confirming 9464 (the OTel Prometheus convention) doesn't collide with anything else expected on target hosts — documented as configurable, so acceptable.

Performance

• Import cost contained. OTel/prometheus SDK imports are deferred into _initialize (called lazily on first activation), and a subprocess test asserts opentelemetry.sdk / opentelemetry.exporter.prometheus are absent from sys.modules when disabled. Consistent with the project's fast-import invariant. The PR notes import-perf/lazy-import checks pass.

• Per-event overhead is minimal. accepts_scheduler_event short-circuits on an empty active-run set before any event construction, and scheduler_event_sink_accepts avoids task_admission_snapshot()/health-diagnostics work when the sink rejects the kind (verified by test_scheduler_filter_skips_snapshot_and_health_diagnostics). Good — this avoids paying telemetry cost on the hot path when disabled.

Testing

• Comprehensive: config bounds/serialization, engine sink forwarding, filtering, ambient correlation, terminal-event exactness, request timing, and interface-level end-to-end scrapes (including concurrent and repeated creates proving cumulative accumulation and exact-once export). Failure-containment paths (port in use, init failure, log-handler failure, idempotent shutdown) are all covered.
• One gap worth noting: no test asserts behavior when two different processes contend for port 9464 (only in-process rebind/reuse). This is inherently hard to test and the _activate_run exception handler covers bind failure (test_runtime_port_failure_does_not_escape), so acceptable.

Docs

• The new "Local OpenTelemetry Metrics" section in architecture-and-performance.mdx is accurate against the implementation: default port, disable semantics, the five instruments, listener lifecycle, the loopback/no-auth warning, and the multiprocessing limitation. The RunConfig.otel_metrics_port docstring and Pydantic description are consistent with each other and with the field bounds (ge=1, le=65535).

Nits

• opentelemetry.py:1298 _non_negative_int — see bool note above.
• data_designer.py — 3× duplicated enable-check ternary; consider a helper.
• dataset_builder.py:976 — getattr where a typed attribute access would do.

Structural Impact

Pre-computed by graphify; annotated with reviewer verification.

Risk: HIGH (1 core abstraction(s) modified; 27 import direction violation(s))

• 20 Python files, 341 AST entities, 7/79 clusters

Import Direction Violations (27)

Legal direction: interface -> engine -> config

• ._is_missing_value() (config) --calls--> .set() (engine)
• column_types() (config) --calls--> .set() (engine)
• generate_analysis_report() (config) --calls--> .set() (engine)
• _validate_skip_scope() (config) --calls--> .set() (engine)
• required_columns() (config) --calls--> .set() (engine)
• +22 more

Core Abstractions Modified

• AsyncTaskScheduler - Add support for images in preview_result.display_sample_record() #5 most connected (146 deps)

High-Connectivity Changes

• AsyncTaskScheduler (146 deps) in packages/data-designer-engine/src/data_designer/engine/dataset_builders/async_scheduler.py
• .set() (116 deps) in packages/data-designer-engine/src/data_designer/engine/observability.py
• DataDesigner (86 deps) in packages/data-designer/src/data_designer/interface/data_designer.py
• DatasetBuilder (73 deps) in packages/data-designer-engine/src/data_designer/engine/dataset_builders/dataset_builder.py
• ResourceProvider (73 deps) in packages/data-designer-engine/src/data_designer/engine/resources/resource_provider.py
• JinjaRenderingEngine (66 deps) in packages/data-designer-config/src/data_designer/config/run_config.py
• RunConfig (59 deps) in packages/data-designer-config/src/data_designer/config/run_config.py
• ModelRequestExecutor (51 deps) in packages/data-designer-engine/src/data_designer/engine/models/clients/model_request_executor.py
• +209 more

Cross-Package Dependencies

• mcp_command() (interface) --calls--> .run() (engine)
• models_command() (interface) --calls--> .run() (engine)
• providers_command() (interface) --calls--> .run() (engine)
• tools_command() (interface) --calls--> .run() (engine)
• .run_create() (interface) --calls--> .model_copy() (config)
• .__init__() (interface) --calls--> .set() (engine)
• +1070 more

Reviewer note on the 27 import-direction violations: these are false positives from call-graph attribution. The flagged config --calls--> .set() (engine) edges resolve to ContextVar.set() (runtime_correlation_provider) and OTel counter/histogram .set()/.add(), not config importing engine. I verified packages/data-designer-config/src/data_designer/config/run_config.py imports nothing from data_designer.engine — the only config change is a plain int | None Pydantic field. No real layering violation is introduced. The AsyncTaskScheduler and observability changes are additive event-sink seams; the OpenTelemetryRuntime is injected into the engine as an anonymous sink so the engine never imports the interface. The HIGH risk rating is driven by the god-node blast radius (scheduler, observability), which warrants the extra scrutiny applied above — but the changes to those nodes are backward-compatible additions, not signature-breaking edits.

Verdict

Approve (with minor, non-blocking nits). This is a high-quality, well-architected PR that respects the interface → engine → config layering, contains all telemetry failures so they cannot alter create results, and has strong privacy guarantees backed by explicit tests. The scheduler run() and request-executor timing refactors preserve their exactly-once emission contracts and re-raise the original exceptions. The structural-analysis HIGH rating reflects god-node connectivity, not a real regression: the flagged import-direction violations are call-graph false positives (ContextVar/OTel .set()), and the god-node edits are backward-compatible additions.

Suggested (optional) follow-ups: dedupe the 3× enable-check ternary in data_designer.py; drop the defensive getattr in dataset_builder.py now that the field is typed; consider guarding _non_negative_int against bool. None block merge.

[greptile-apps]

Greptile Summary

This PR adds pull-based OpenTelemetry/Prometheus metrics for DataDesigner.create() jobs, exposing seven instruments (create duration, dataset records/progress, scheduler events, active model requests, GenAI operation duration, and log record counts) on a loopback HTTP endpoint controlled by a new RunConfig.otel_metrics_port field (default 9464, None to disable).

• A new OpenTelemetryRuntime singleton owns the meter provider, Prometheus registry, and HTTP listener; it is wired into DataDesigner via a per-run observe_create context manager that sets a RuntimeCorrelation context variable, registers the run, and guarantees gauge reconciliation in a finally block.
• The engine-layer SchedulerAdmissionEventSink and RequestAdmissionEventSink protocols gain cheap interest-filtering via accepts_scheduler_event/accepts_request_event, and a new _FanoutSchedulerEventSink lets the existing JSONL sink and the OTel sink coexist through exception-isolated fan-out.
• ModelRequestExecutor._execute_sync/async are refactored to use a single finally block for the model_request_completed emit, capturing request duration correctly across all outcome paths, and AsyncTaskScheduler emits scheduler_job_completed in a finally block that covers success, early_shutdown, cancellation, and error outcomes.

Confidence Score: 5/5

Safe to merge — the telemetry layer is fully isolated, all terminal paths emit correctly, and the previously-flagged deadlock on port rebind is resolved by calling _stop_server outside the lock.

The implementation is careful about concurrency: RLock is used for reentrant acquisition from emit_scheduler_event into _finish_active_dataset/_advance_dataset_progress, _stop_server is invoked after releasing the lock so a concurrent Prometheus scrape cannot deadlock a port rebind, and the observe_create finally block reconciles any leaked gauge increments from interrupted requests. The scheduler_job_completed refactor correctly covers all outcome paths (success, early_shutdown, cancelled, error) through a single finally block, and the model_request_completed timing refactor captures duration accurately across every exception branch.

No files require special attention.

Important Files Changed

Filename	Overview
packages/data-designer/src/data_designer/integrations/opentelemetry.py	New file: implements OpenTelemetryRuntime with per-run lifecycle management, gauge reconciliation, and Prometheus HTTP listener; _stop_server is correctly called outside the RLock to avoid the scrape-handler deadlock raised in a previous review thread.
packages/data-designer-engine/src/data_designer/engine/observability.py	Adds SchedulerAdmissionEventKind/RequestAdmissionEventKind type aliases, scheduler_event_sink_accepts/request_event_sink_accepts interest-filtering helpers, and a _FanoutSchedulerEventSink that isolates per-sink failures; logic is correct and well-tested.
packages/data-designer-engine/src/data_designer/engine/dataset_builders/async_scheduler.py	Moves scheduler_job_completed into a finally block covering all terminal outcomes (success, early_shutdown, cancelled, error); health diagnostics are attached only for non-error outcomes to avoid expensive snapshots on abnormal exits.
packages/data-designer-engine/src/data_designer/engine/models/clients/model_request_executor.py	Refactors _execute_sync and _execute_async to emit model_request_completed in a single finally block with accurate duration_seconds; all exception paths (ProviderError, TimeoutError, CancelledError, BaseException) correctly set outcome before the finally runs.
packages/data-designer/src/data_designer/interface/data_designer.py	Wires OpenTelemetryRuntime into DataDesigner via an _observe_create decorator on create(); scheduler and request event sinks are consistently gated on otel_metrics_port is not None in both _create_resource_provider and _create_request_admission_controller.
packages/data-designer-config/src/data_designer/config/run_config.py	Adds otel_metrics_port: int
packages/data-designer-engine/src/data_designer/engine/resources/resource_provider.py	Adds scheduler_event_sink and request_event_sink parameters to create_resource_provider; the scheduler_event_sink is stored on ResourceProvider and fan-out composed with the JSONL sink in DatasetBuilder.
packages/data-designer/tests/integrations/test_opentelemetry.py	New test file with ~996 lines covering runtime lifecycle, event filtering, concurrent runs, port rebind, progress tracking, log handler, and integration with a real HTTP scrape; comprehensive coverage of the new OTel integration.

Sequence Diagram

%%{init: {'theme': 'neutral'}}%%
sequenceDiagram
    participant User
    participant DataDesigner
    participant OTelRuntime as OpenTelemetryRuntime
    participant Scheduler as AsyncTaskScheduler
    participant Executor as ModelRequestExecutor
    participant Prometheus

    User->>DataDesigner: create(config_builder)
    DataDesigner->>OTelRuntime: observe_create(port)
    OTelRuntime->>OTelRuntime: _activate_run(run_id, port) - start HTTP server outside lock
    OTelRuntime-->>DataDesigner: run_id in _active_run_ids

    DataDesigner->>Scheduler: build(num_records)
    Scheduler->>OTelRuntime: emit_scheduler_event(scheduler_job_started)
    OTelRuntime->>OTelRuntime: _start_dataset_progress(run_id, total)

    loop per row-group
        Scheduler->>Executor: execute request
        Executor->>OTelRuntime: emit_request_event(model_request_started)
        OTelRuntime->>OTelRuntime: _active_model_requests.add(+1)
        Executor->>Executor: call() - LLM
        Executor->>OTelRuntime: emit_request_event(model_request_completed, duration)
        OTelRuntime->>OTelRuntime: _active_model_requests.add(-1) + _request_duration.record()
        Scheduler->>OTelRuntime: emit_scheduler_event(row_group_checkpointed)
        OTelRuntime->>OTelRuntime: _advance_dataset_progress(run_id)
        Prometheus->>OTelRuntime: GET /metrics
        OTelRuntime-->>Prometheus: progress, active_requests, counters
    end

    Scheduler->>OTelRuntime: emit_scheduler_event(scheduler_job_completed)
    OTelRuntime->>OTelRuntime: _finish_active_dataset(run_id)

    DataDesigner->>OTelRuntime: observe_create exits (finally)
    OTelRuntime->>OTelRuntime: discard run_id + _finish_active_requests reconciliation
    OTelRuntime->>OTelRuntime: _record_create_duration
    Note over OTelRuntime,Prometheus: HTTP server stays up for final scrape

Loading

%%{init: {'theme': 'base', 'themeVariables': {"darkMode": true, "background": "#0d1117", "primaryColor": "#21262d", "primaryTextColor": "#e6edf3", "primaryBorderColor": "#8b949e", "lineColor": "#8b949e", "textColor": "#e6edf3", "edgeLabelBackground": "#161b22", "actorBkg": "#21262d", "actorBorder": "#8b949e", "actorTextColor": "#e6edf3", "actorLineColor": "#8b949e", "signalColor": "#8b949e", "signalTextColor": "#e6edf3", "noteBkgColor": "#373320", "noteBorderColor": "#d4a72c", "noteTextColor": "#f0e6c0", "labelBoxBkgColor": "#21262d", "labelBoxBorderColor": "#8b949e", "labelTextColor": "#e6edf3", "loopTextColor": "#e6edf3", "activationBkgColor": "#30363d", "activationBorderColor": "#8b949e"}}}%%
sequenceDiagram
    participant User
    participant DataDesigner
    participant OTelRuntime as OpenTelemetryRuntime
    participant Scheduler as AsyncTaskScheduler
    participant Executor as ModelRequestExecutor
    participant Prometheus

    User->>DataDesigner: create(config_builder)
    DataDesigner->>OTelRuntime: observe_create(port)
    OTelRuntime->>OTelRuntime: _activate_run(run_id, port) - start HTTP server outside lock
    OTelRuntime-->>DataDesigner: run_id in _active_run_ids

    DataDesigner->>Scheduler: build(num_records)
    Scheduler->>OTelRuntime: emit_scheduler_event(scheduler_job_started)
    OTelRuntime->>OTelRuntime: _start_dataset_progress(run_id, total)

    loop per row-group
        Scheduler->>Executor: execute request
        Executor->>OTelRuntime: emit_request_event(model_request_started)
        OTelRuntime->>OTelRuntime: _active_model_requests.add(+1)
        Executor->>Executor: call() - LLM
        Executor->>OTelRuntime: emit_request_event(model_request_completed, duration)
        OTelRuntime->>OTelRuntime: _active_model_requests.add(-1) + _request_duration.record()
        Scheduler->>OTelRuntime: emit_scheduler_event(row_group_checkpointed)
        OTelRuntime->>OTelRuntime: _advance_dataset_progress(run_id)
        Prometheus->>OTelRuntime: GET /metrics
        OTelRuntime-->>Prometheus: progress, active_requests, counters
    end

    Scheduler->>OTelRuntime: emit_scheduler_event(scheduler_job_completed)
    OTelRuntime->>OTelRuntime: _finish_active_dataset(run_id)

    DataDesigner->>OTelRuntime: observe_create exits (finally)
    OTelRuntime->>OTelRuntime: discard run_id + _finish_active_requests reconciliation
    OTelRuntime->>OTelRuntime: _record_create_duration
    Note over OTelRuntime,Prometheus: HTTP server stays up for final scrape

Loading

_{Reviews (6): Last reviewed commit: "fix: harden OpenTelemetry runtime" | Re-trigger Greptile}

[andreatgretel]

One inline finding.

[andreatgretel]

[P2] Keep started-event emission inside the lease cleanup scope

The admission lease is already acquired here, but this emission occurs before the try that guarantees its release. If KeyboardInterrupt occurs while the sink is running, the coroutine exits with the lease still active. I reproduced active_lease_count == 1 on this head, while the same forced path on main releases it with the local_cancelled outcome. Initialize the timing and outcome variables first, then emit inside the protected try. The synchronous attempt has the same regression.

[eric-tramel]

This is fixed on the current head as well. Started-event delivery now occurs inside the cleanup scope for both sync and async attempts, so a BaseException during sink delivery still closes the acquired admission lease with the correct terminal outcome. The regression tests force that path and verify the original exception is preserved and the active lease count returns to zero. Current full validation: 3,940 passing tests, one existing skip.

[nabinchha]

Nice work on this one, @eric-tramel — this is a well-structured, privacy-conscious observability layer with impressively thorough tests.

Summary

Adds default, pull-based OpenTelemetry/Prometheus metrics for DataDesigner.create() on a loopback endpoint controlled by RunConfig.otel_metrics_port (default 9464, None to disable). The interface package owns the OTel SDK, meter provider, and HTTP listener; existing engine scheduler/request event seams feed the exporter, and a metric-only log handler tees severity counts. The implementation matches the PR description, keeps the config → engine → interface layering intact, defers heavy imports into _initialize(), and contains telemetry failures so they can't change create results.

Findings

Warnings — Worth addressing

packages/data-designer/src/data_designer/interface/data_designer.py:173 + run_config.py:194 — Telemetry is on by default, opening a listening socket on every create()

• What: otel_metrics_port defaults to 9464, so every DataDesigner.create() in every environment now binds a TCP listener (loopback) and keeps it open until process exit. This is a new, always-on side effect for existing users who never asked for metrics.
• Why: In multi-instance hosts, CI, or notebook workflows this can surprise operators, and security scanners routinely flag open ports even on loopback. The graceful-degradation path (bind failure → warn → continue) is good, but the default posture is still "open a socket unless told not to."
• Suggestion: Worth a quick team confirmation that default-on is the intended contract vs. opt-in (e.g. default None, enabled explicitly). If default-on is deliberate, consider calling it out prominently in the release notes/changelog so operators aren't caught off guard. Not blocking if the team already signed off on this.

packages/data-designer-engine/src/data_designer/engine/models/clients/model_request_executor.py:125,176 — Confirming the existing P2 (started-event emitted before the release-guarding try)

• What: I'm not re-raising new feedback here, just corroborating @andreatgretel's inline comment: _emit_model_event("model_request_started", ...) runs after the lease is acquired but before the try that guarantees release. _emit_model_event swallows Exception but not BaseException, so a KeyboardInterrupt during the started emit leaks the lease. This still reproduces on the current head.
• Why: It's a genuine (if narrow) resource-leak window, and the fix is cheap.
• Suggestion: Move the started emission inside the protected try (initialize started_at/duration_seconds/outcome first, then emit), as suggested in the inline thread. Same for both the sync and async attempt paths.

packages/data-designer-engine/src/data_designer/engine/models/clients/model_request_executor.py:272 — Request correlation relies on the contextvar being present on the calling thread

• What: _item() reads runtime_correlation_provider.current(), and emit_request_event drops any event whose captured_correlation has no run_id. The active-request gauge and gen_ai.client.operation.duration therefore only populate when the model call executes in a context that inherited the contextvar set by observe_create.
• Why: This is fine for the async path (tasks inherit the context), but for any sync/thread-pool execution mode that runs client calls on worker threads that didn't copy the context, current() returns None and those requests silently won't appear in the model-activity metrics. Want to make sure that's understood/acceptable.
• Suggestion: Could we confirm the concurrency modes that route through ModelRequestExecutor all preserve the contextvar (or add a targeted test for the sync/threaded path)? If sync-mode gaps are expected, a one-line note in the docs/metric description would set expectations.

packages/data-designer/src/data_designer/integrations/opentelemetry.py — OpenTelemetryRuntime is carrying four responsibilities (SRP)

• What: One ~500-line class owns (a) HTTP server + meter-provider lifecycle, (b) instrument definitions, (c) a domain state machine (active runs, dataset-progress math, in-flight request accounting), and (d) event adaptation for three sinks. The progress aggregation (_start_dataset_progress/_advance_dataset_progress/_finish_active_dataset/_active_dataset_progress, ~404-435) is a self-contained, OTel-agnostic state machine.
• Why: The progress semantics can only be tested by standing up a Prometheus server, and every concern shares the same _lock. This will get harder to reason about as instruments are added.
• Suggestion: Pull the progress math into a small DatasetProgressTracker (no OTel deps, directly unit-testable) and consider a frozen _MetricInstruments container for the instruments; leave OpenTelemetryRuntime as orchestration + lifecycle. Fine as a follow-up if it's too much for this PR. (Naming: DatasetProgressTracker stays un-prefixed since the whole point is direct testability — STYLEGUIDE prefers public over private for testable helpers — whereas _MetricInstruments is a dumb container tested through the runtime, so the underscore fits.)

packages/data-designer/src/data_designer/integrations/opentelemetry.py:84-96 — ~11 attributes typed Any sidestep the "typed code" invariant

• What: _registry, _meter_provider, _server, _create_duration, _dataset_records, … are all Any because the OTel types aren't imported at module scope.
• Why: AGENTS.md/STYLEGUIDE call out avoiding overly broad Any; as-is the checker can't catch e.g. calling .add() on a histogram vs .record() on a counter.
• Suggestion: Import the OTel/prometheus types under TYPE_CHECKING (the same pattern used for lazy heavy libs) and annotate precisely (Histogram, Counter, UpDownCounter, ObservableGauge, MeterProvider, CollectorRegistry, HTTPServer). This is a low-risk, high-clarity win and pairs naturally with the split above.

model_request_executor.py:287 + data_designer.py:745 — Hot-path events are built before the gate, and preview() pays for events it always drops

• What: _emit_model_event unconditionally builds the full RequestAdmissionEvent — including self._request_admission.pressure.snapshot(...) and the recursive _json_safe normalization in __post_init__ — whenever _event_sink is not None, then dispatches to emit_request_event, which drops it if the run isn't active. Unlike the scheduler path there's no accepts_request_event fast-path filter. Because _create_resource_provider wires the sink for both create and preview (only create activates a run), every model call on the preview path does this build-then-drop.
• Why: Model requests are the highest-volume events, and preview is latency-sensitive; the per-call snapshot + JSON normalization is wasted work there.
• Suggestion: Add an accepts_request_event(event_kind) gate mirroring scheduler_event_sink_accepts and check it before building the event / taking the pressure snapshot. That both restores symmetry with the scheduler path and makes the inactive-run/preview case exit cheaply (cleaner than special-casing preview at the wiring site).

Engine observability now has three inconsistent emission seams (strategic design debt) — observability.py, request_admission/controller.py:770-777, async_scheduler.py, model_request_executor.py

• What: The DIP itself is clean (engine defines the sink Protocols; the interface's OpenTelemetryRuntime implements them; correlation flows the other way via the runtime_correlation_provider contextvar). But the emission side now has three different styles side by side:
  • AsyncScheduler → single direct sink with an accepts_scheduler_event interest filter.
  • RequestAdmissionController._emit_events → direct sink and the global emit_request_admission_event callback registry (the latter is how progress/reporter.py consumes events), no filter.
  • ModelRequestExecutor → single direct sink only (so model_request_started/completed reach OTel but not the progress reporter), no filter.
• Why: There's no single answer to "how do I observe request activity?" The controller's dual fan-out also has mismatched reliability — the direct-sink call is try/except-guarded but the emit_request_admission_event(event) call on line 777 is not. As more consumers appear, this divergence compounds.
• Suggestion: Converge on one seam over time — e.g. make the progress reporter just another registered subscriber/sink, and give both scheduler and request paths the same optional accepts_* filter, so the model is uniformly "producer emits event → fan-out to N sinks." Bigger than this PR should take on, but worth capturing as the design debt this feature adds.

packages/data-designer-engine/src/data_designer/engine/observability.py:227 — accepts_scheduler_event is a phantom (undeclared) protocol method

• What: scheduler_event_sink_accepts probes getattr(sink, "accepts_scheduler_event", None), but that method isn't on the SchedulerAdmissionEventSink Protocol (which only declares emit_scheduler_event).
• Why: This is a latent bug, not just a style nit — a rename/typo of the method silently degrades to "accept everything" (the filter becomes a no-op), and neither the type checker nor the contract would catch it. Given the filter exists partly for hot-path cost, a silent no-op has real consequences.
• Suggestion: Declare accepts_scheduler_event on the Protocol (or a FilteringSchedulerAdmissionEventSink sub-Protocol checked via isinstance) so the filter is part of the contract instead of duck-typed.

packages/data-designer/src/data_designer/integrations/opentelemetry.py:66-70 — Swallowed exceptions make telemetry bugs undiagnosable (error handling)

• What: The "never break create" policy is applied via bare swallows — e.g. _MetricLogHandler.emit does except Exception: pass with no logging at all — so a genuine bug inside the instrumentation is completely invisible in the field.
• Why: STYLEGUIDE flags swallowed exceptions as an error-handling problem. The intent (don't break create) is right, but "silently drop with zero trace" means a broken metric would never be noticed or reproducible.
• Suggestion: Keep the swallow, but log at debug with exc_info=True instead of bare pass, and optionally gate a strict re-raise behind an env var (e.g. DATA_DESIGNER_OTEL_STRICT=1) for tests/field debugging.

packages/data-designer-config/src/data_designer/config/run_config.py:194 — otel_metrics_port=None reads like a kill switch but isn't

• What: None means "don't activate for this invocation," yet a listener started by a prior run keeps serving. The field overloads "enabled" and "which port."
• Why: This is a user-facing footgun with a security angle — someone who sets None specifically to stop metrics being served will still have the prior run's endpoint open. Combined with the default-on Warning above, the mismatch between expectation and behavior could bite.
• Suggestion: At minimum, clarify in the docstring that None disables activation for the invocation only (not a process-wide stop), and document how to actually stop the listener. Splitting into enabled: bool + port: int would remove the overload but is probably more than needed.

Suggestions — Take it or leave it

Four optional polish items (click to expand)

packages/data-designer-engine/src/data_designer/engine/dataset_builders/dataset_builder.py:976 — Defensive getattr on a field that always exists

• What: scheduler_event_sink=getattr(self._resource_provider, "scheduler_event_sink", None). Now that ResourceProvider declares scheduler_event_sink: SchedulerAdmissionEventSink | None = None, the attribute always exists.
• Why: The getattr fallback reads as "this might not be there," which is misleading for a declared Pydantic field.
• Suggestion: Simplify to self._resource_provider.scheduler_event_sink unless there's a compat reason (e.g. externally-constructed providers) — in which case a short comment would help.

packages/data-designer/src/data_designer/interface/data_designer.py:743-757 — The "is telemetry on?" rule is duplicated four times (DRY)

• What: self._open_telemetry if self._run_config.otel_metrics_port is not None else None appears at three call sites (request sink, scheduler sink, request-admission controller), plus the port passed into observe_create.
• Why: The enablement invariant is smeared across the interface; changing it means touching several spots.
• Suggestion: Extract a self._metrics_sink() accessor returning the runtime or None, and use it at each site.

packages/data-designer/src/data_designer/integrations/opentelemetry.py:491-492 — Global singleton + atexit registered as an import side effect

• What: Importing the module constructs _RUNTIME and registers atexit.register(_RUNTIME.shutdown).
• Why: A process-global lifecycle established just by importing the interface is a classic testability smell (the tests already work around it with a fresh-instance fixture). It's cheap today (no OTel imports fire at construction), so low urgency.
• Suggestion: Optionally make get_open_telemetry_runtime() construct lazily on first call and register atexit there once.

packages/data-designer/src/data_designer/integrations/opentelemetry.py:335-345 — Progress gauge blends independent jobs into one process-wide float

• What: _dataset_progress_value is a single number aggregated across all active runs, with the last finished value lingering until the next start.
• Why: For concurrent creates, one gauge that shifts between "this job / blended / stale residual" is semantically muddy on a dashboard. Fine for the common single-job case.
• Suggestion: If overlapping creates become normal, prefer a per-run gauge label so each job reports its own fraction. Defer to a follow-up unless multi-job dashboards are an immediate need (it's a metric-schema change).

What Looks Good

• Failure containment is excellent — _activate_run, _initialize, the log handler, _record_create_duration, and _finish_active_requests all defensively suppress/rollback so telemetry can never break a create. The partial-init cleanup in _initialize (removing the handler and shutting down the provider on failure) is especially careful.
• The P1 deadlock fix is done right — _rebind now returns the old (server, thread) and _stop_server runs outside self._lock in both _activate_run and shutdown, with test_runtime_stops_rebound_listener_outside_runtime_lock locking in the behavior. The RLock choice for the re-entrant progress helpers is appropriate.
• Privacy posture is thorough — _bounded_attribute strips/bounds attribute values, and the metric-only log handler tees severity counts without ever exporting log bodies. The docs section spells out the loopback-only, no-auth, metrics-only contract clearly.
• Test suite is genuinely comprehensive — post-teardown event rejection, orphan/duplicate request reconciliation, idle-listener rebind, conflicting-port reuse, concurrent creates sharing one exporter, and test_disabled_runtime_does_not_import_otel_sdk_in_fresh_process guarding the lazy-import contract. These verify behavior, not plumbing.

Suggested Follow-up Issues

For the larger design items we're intentionally not folding into this PR, let's open tracking issues so they don't get lost (happy to file these — just confirm):

• Unify the three engine observability emission seams — converge direct-sink / global-registry / sink-with-filter into one "producer → fan-out to N sinks" model, including guarding the unguarded emit_request_admission_event call and giving both paths an accepts_* filter. (Strategic; highest-value.)
• Split OpenTelemetryRuntime — extract DatasetProgressTracker (directly unit-testable) and a typed _MetricInstruments container; leave the runtime as orchestration + lifecycle.
• Lazy-construct the runtime singleton — move construction + atexit registration into get_open_telemetry_runtime() to drop the import side effect.
• Per-run progress gauge label — only if overlapping creates become a supported scenario (metric-schema change).

The Warnings above are small enough to fix in this PR rather than defer (the accepts_scheduler_event Protocol declaration, the swallowed-exception logging, and the otel_metrics_port=None docstring), as are the DRY _metrics_sink() accessor and the getattr cleanup.

Verdict

Needs changes — nothing here is a hard blocker on the happy path, but before merge it's worth: (1) getting explicit sign-off on default-on telemetry (or flipping to opt-in), and (2) closing the still-open P2 lease-leak window in model_request_executor. The correlation-propagation question is a "confirm or add a test" item.

On the design side, the low-risk Warnings worth folding into this PR are the hot-path/preview build-then-drop (accepts_request_event gate), the Any typing (TYPE_CHECKING), the phantom-protocol method (declare accepts_scheduler_event), the swallowed-exception logging, and the otel_metrics_port=None docstring — plus the small _metrics_sink()/getattr cleanups. The god-object split, lazy singleton, per-run progress label, and especially the three inconsistent emission seams are the larger items better tracked as follow-up issues than jammed into this PR.

ruff check and ruff format --check pass clean on all changed source files.

---

This review was generated by an AI assistant.

[andreatgretel]

One inline question.

[eric-tramel]

Implemented the requested review follow-ups in 8b63e675 and merged the latest main into the branch.

What changed:

• Moved sync and async model_request_started delivery inside the lease cleanup scope, with regressions proving KeyboardInterrupt cannot leak a lease.
• Added optional request-event interest filtering in both the model executor and admission controller, so inactive/preview OTel paths skip event delivery and attempt snapshot construction.
• Strengthened runtime-correlation coverage across the outer asyncio.to_thread hop and the sync-to-async engine bridge.
• Preserved scheduler completion health diagnostics and report early shutdown as outcome="early_shutdown" rather than success.
• Bounded exported model labels to 128 characters while preserving valid / separators.
• Replaced the defensive scheduler-sink getattr with the declared ResourceProvider field.

Validation:

• make check-all passed.
• make test passed: 3,907 passed, 1 skipped.
• Focused scheduler/request/OTel regression suite: 302 passed.
• Seven-persona delegated review completed; targeted follow-ups report no remaining actionable findings.
• GitHub checks are complete: 60 passed, 2 expected skips, 0 failures.
• Greptile re-reviewed 8b63e675 at 5/5 and reports “Safe to merge” with no new findings.

The PR is conflict-free and mergeable. Please re-review/clear the prior changes-requested state when convenient.

[eric-tramel]

Rebased this PR onto current main (7798a9b8) and safely force-pushed the rewritten head at eeae00e0.

Meaningful rebase resolution:

• Preserved main's per-run JSONL scheduler event sink and the OTel provider sink by composing them through a generic, exception-isolated fan-out. Both observers now receive events when enabled; scheduler/task code still has no OTel dependency.
• Preserved the intervening max_concurrent_row_groups, write_scheduler_events, and auto_configure_logging behavior from main.
• Added a builder-level regression proving that one scheduler event reaches both configured sinks.
• Removed the obsolete merge commit; the branch is now six commits directly atop main and zero commits behind.

Validation on the final tree:

• make check-all passed.
• make test passed: 3,939 passed, 1 existing skip.
• Focused rebase/OTel suite passed: 410 passed.
• Fern check passed: 0 errors, 1 existing warning.
• Three independent local review lanes (correctness, API compatibility, and test completeness) found no remaining actionable issues.
• GitHub checks completed: 60 passed, 2 expected skips, 0 failures.
• Greptile reviewed eeae00e0 at 5/5, reports “Safe to merge,” and found no new issues.

The prior lease-cleanup P2 thread is outdated and remains covered by the current fix/regressions. The PR is conflict-free and GitHub reports it as mergeable; the only remaining merge gate is the existing CHANGES_REQUESTED review state. @nabinchha @andreatgretel, please re-review/clear that stale gate when convenient.

[andreatgretel]

There's a second deadlock in the same family as the P1 that was fixed: this logger.warning runs while holding self._lock, and so do the logger.info in _rebind and the warning in _finish_active_requests. The trouble is _MetricLogHandler sits on the data_designer logger, and Python's logging holds a per-handler lock around emit(), which calls record_log, which takes self._lock. So a normally-logging engine thread takes handler lock then runtime lock, while these paths take runtime lock then handler lock. Two threads interleaving those orders hang each other, and since one of them keeps the handler lock, every data_designer log call in the process hangs after that too, and the process can't even exit (logging.shutdown() blocks at interpreter exit). Concurrent creates open the window, e.g. a second create with a different port hits this exact line while the first one's workers are logging.

Claude Code and Codex both reproduced this with two threads. The fix is the same shape as the P1: never log while holding the lock (capture the message, log after release), and ideally make record_log's gate a lock-free read so the logging path never touches the runtime lock at all.

[eric-tramel]

Fixed in 9aee1dde. The three runtime-owned logging paths now collect messages/errors while _lock is held and call logger.* only after the outer lock is released: initial/reused bind logging, the conflicting-port warning, and the active-request reconciliation warning. I kept record_log under the lock because its gate must remain ordered with teardown; making that read lock-free would reintroduce stale post-teardown metrics. I added a deterministic regression covering all three callbacks and reproduced the real handler-lock inversion; after this fix, handler-blocked logging no longer holds the runtime lock. The focused OTel suite has 31 passing tests and the full suite has 3,940 passing tests (one existing skip).

[andreatgretel]

packages/data-designer-engine/src/data_designer/engine/dataset_builders/dataset_builder.py:976

scheduler_event_sink=self._resource_provider.scheduler_event_sink,

heads up: #800 landed on main about half an hour after this branch's last commit, and it wires JsonlSchedulerEventSink into the same single scheduler_event_sink slot this PR uses for the OTel runtime. The branch now conflicts with main on dataset_builder.py, observability.py, and a few test files, so the "conflict-free and mergeable" state is stale. The tricky part is that the resolution is semantic, not textual: with one slot, picking either side silently drops the other feature's events (no JSONL persistence, or no OTel scheduler metrics). A small fan-out sink during the rebase (emit to both, guarded per sink, accepts = OR of the members) would let them coexist, with a test for JSONL-only / OTel-only / both.

One more thing worth flagging to the #800 folks: this PR moves scheduler_job_completed into a finally, so after the merge, JSONL event logs will also contain terminal events for cancelled and failed runs, with slimmer diagnostics than success events.

[andreatgretel]

Codex flagged this one during cross-review: stripping characters before truncating isn't collision-resistant, so distinct models can merge into one Prometheus series. llama3:8b and llama38b both normalize to llama38b, and two model IDs that differ only past character 128 collide too. Since these labels define the series for the active-request gauge and the duration histogram, the dashboards would silently blend latency and concurrency from different endpoints. Prometheus and OTel allow arbitrary strings as label values, so I think the length cap alone gives you the bounding, maybe with a stable hash suffix appended whenever the value had to be modified.

[eric-tramel]

Fixed in 9aee1dde. Short provider/model identities are now preserved verbatim, so punctuation cannot collapse distinct series. Values over 128 characters retain a readable prefix plus a stable 16-hex SHA-256 suffix, preserving the existing length bound without same-prefix collisions. This is applied to both gen_ai.provider.name and gen_ai.request.model, with regressions for llama3:8b versus llama38b and for two long same-prefix model IDs. The focused OTel suite has 31 passing tests and the full suite has 3,940 passing tests (one existing skip).

[eric-tramel]

@andreatgretel — points 1 and 2 are addressed in 9aee1dde, with details and validation posted on each inline thread.

• Lock ordering: runtime-owned lifecycle/reconciliation logs are emitted only after _lock is released. record_log remains locked so its admission check stays serialized with teardown. A deterministic regression exercises all three former lock-inversion sites, and the real handler-lock reproduction now confirms the runtime lock remains available.
• Metric identity: exact short provider/model values are preserved; long values remain bounded to 128 characters with a stable 16-hex SHA-256 suffix. Regressions cover punctuation collisions and long same-prefix identities.
• I also rechecked the earlier lease-cleanup fix and replied on that thread. The current scheduler observer path still exception-isolates and fans out to both per-run JSONL and OTel sinks; the terminal JSONL events are intentional, not duplicate instrumentation.

I ran the requested seven-persona review (elegance, idiomatic Python, correctness, completeness, security, YAGNI, and scope). No actionable findings were accepted. The only suggestion was a P4 YAGNI proposal to return unbounded raw label values; I rejected it because the PR explicitly guarantees bounded metric attributes, and hash-suffixed truncation preserves that contract without merging distinct identities. The patch stays scoped to the OTel runtime and its tests—no dependency, API, or layering changes.

Validation is clean: make check-all, 31 focused OTel tests, and make test with 3,940 passing tests plus one existing skip. The PR description is updated. I have left the review threads unresolved for your verification and would appreciate a re-review when convenient.