datarobot_genai.core.telemetry.datarobot_otel

datarobot_otel

DataRobot OTel ingest endpoint helpers.

Three responsibilities, intentionally co-located so a single import covers both the NAT telemetry exporter (datarobot_genai.dragent.plugins.datarobot_otelcollector) and the framework-instrumentor bootstrap (datarobot_genai.core.telemetry.agent):

• resolve_*_from_env — read MLOPS_DEPLOYMENT_ID / WORKLOAD_ID / DATAROBOT_API_TOKEN / DATAROBOT_(PUBLIC_)ENDPOINT and shape them into the values the OTel ingest expects (deployment-<id> or workload-<id> entity id; <host>/otel/v1/traces or <host>/otel/v1/metrics endpoint).
• bootstrap_otel_provider_for_datarobot — install a global OTel SDK TracerProvider pointed at the DataRobot ingest so framework auto-instrumentors actually export spans.
• bootstrap_otel_meter_provider_for_datarobot — same, for metrics: without it, datarobot_dome (moderations) guard metrics record through the OTel SDK's no-op default MeterProvider and are silently dropped, since nothing else in this SDK's bootstrap path ever installs a real one.

bootstrap_otel_provider_for_datarobot

bootstrap_otel_provider_for_datarobot() -> bool

Ensure framework auto-instrumentor spans reach the DataRobot OTel ingest.

Sibling to (not replacing) the NAT-side datarobot_otelcollector exporter: NAT pipes its own IntermediateStep-derived spans through its own channel; this path serves framework auto-instrumentors that emit spans through the OTel SDK's standard trace.get_tracer(...) API.

Two modes, picked at call time based on what's already in the global TracerProvider slot:

• install — slot still holds the default ProxyTracerProvider: we create a new SDK TracerProvider with the DataRobot exporter and set it globally.
• attach — slot already holds an SDK TracerProvider (e.g. the dragent_fastapi server's startup telemetry layer installs one before NAT plugin discovery happens): we add our own span processor to it instead of replacing it. The pre-existing provider's resource (including its service.name) is kept — DataRobot's OTel ingest routes off the X-DataRobot-* headers we set on the exporter, not off resource attributes, so the merge is safe.

Entity identity is derived from MLOPS_DEPLOYMENT_ID or WORKLOAD_ID (deployment takes precedence).

Returns True when a processor was installed or attached by this call, False (silently) when:

• the hosted-runtime env is incomplete (MLOPS_DEPLOYMENT_ID or WORKLOAD_ID, DATAROBOT_API_TOKEN, or DATAROBOT_(PUBLIC_)ENDPOINT missing) — the local-dev / CI shape;
• something other than an SDK TracerProvider or the default proxy is already installed (we can't attach to an unknown provider type);
• this function has already run successfully in this process.

Source code in datarobot_genai/core/telemetry/datarobot_otel.py

def bootstrap_otel_provider_for_datarobot() -> bool:
    """Ensure framework auto-instrumentor spans reach the DataRobot OTel ingest.

    Sibling to (not replacing) the NAT-side ``datarobot_otelcollector`` exporter:
    NAT pipes its own ``IntermediateStep``-derived spans through its own
    channel; this path serves framework auto-instrumentors that emit spans
    through the OTel SDK's standard ``trace.get_tracer(...)`` API.

    Two modes, picked at call time based on what's already in the global
    ``TracerProvider`` slot:

    * **install** — slot still holds the default ``ProxyTracerProvider``: we
      create a new SDK ``TracerProvider`` with the DataRobot exporter and set
      it globally.
    * **attach** — slot already holds an SDK ``TracerProvider`` (e.g. the
      ``dragent_fastapi`` server's startup telemetry layer installs one before
      NAT plugin discovery happens): we add our own span processor to it instead
      of replacing it. The pre-existing provider's resource
      (including its ``service.name``) is kept — DataRobot's OTel ingest
      routes off the ``X-DataRobot-*`` headers we set on the exporter, not off
      resource attributes, so the merge is safe.

    Entity identity is derived from ``MLOPS_DEPLOYMENT_ID`` or ``WORKLOAD_ID``
    (deployment takes precedence).

    Returns ``True`` when a processor was installed or attached by this call,
    ``False`` (silently) when:

    * the hosted-runtime env is incomplete (``MLOPS_DEPLOYMENT_ID`` or
      ``WORKLOAD_ID``, ``DATAROBOT_API_TOKEN``, or
      ``DATAROBOT_(PUBLIC_)ENDPOINT`` missing) — the local-dev / CI shape;
    * something other than an SDK ``TracerProvider`` or the default proxy is
      already installed (we can't attach to an unknown provider type);
    * this function has already run successfully in this process.
    """
    if _BOOTSTRAP_STATE["installed"]:
        return False

    headers = resolve_datarobot_headers_from_env()
    endpoint = resolve_otel_traces_endpoint_from_env()
    if not headers or not endpoint:
        logger.info(
            "Skipping OTel TracerProvider bootstrap: hosted-runtime env "
            "(MLOPS_DEPLOYMENT_ID or WORKLOAD_ID / DATAROBOT_API_TOKEN / "
            "DATAROBOT_(PUBLIC_)ENDPOINT) not fully set."
        )
        return False

    # Imported lazily so test environments without the SDK don't pay the cost
    # at module import. All three packages are pinned by the dragent extra.
    from opentelemetry import trace
    from opentelemetry.exporter.otlp.proto.http.trace_exporter import OTLPSpanExporter
    from opentelemetry.sdk.resources import Resource
    from opentelemetry.sdk.trace import TracerProvider
    from opentelemetry.sdk.trace.export import BatchSpanProcessor
    from opentelemetry.sdk.trace.export import SimpleSpanProcessor
    from opentelemetry.trace import ProxyTracerProvider

    try:
        # Inspect the global slot before building anything: a BatchSpanProcessor
        # spawns a worker thread (and the exporter an HTTP session) at
        # construction, so we must not build them on the skip path below.
        current = trace.get_tracer_provider()
        if not isinstance(current, (ProxyTracerProvider, TracerProvider)):
            logger.debug(
                "Skipping OTel TracerProvider bootstrap: non-SDK provider "
                "already installed (%s); cannot attach a span processor.",
                type(current).__name__,
            )
            return False

        exporter = OTLPSpanExporter(
            endpoint=endpoint,
            headers=headers,
        )
        processor = (
            SimpleSpanProcessor(exporter)
            if _use_simple_span_processor()
            else BatchSpanProcessor(exporter)
        )

        from .nat_tracer import wrap_sdk_tracer_provider

        if isinstance(current, ProxyTracerProvider):
            sdk_version = _get_opentelemetry_sdk_version()
            resource = Resource.create(
                {
                    "telemetry.sdk.language": "python",
                    "telemetry.sdk.name": "opentelemetry",
                    "telemetry.sdk.version": sdk_version,
                    "service.name": _resolve_service_name(),
                }
            )
            provider = wrap_sdk_tracer_provider(TracerProvider(resource=resource))
            provider.add_span_processor(processor)
            trace.set_tracer_provider(provider)
            action = "installed"
        else:  # SDK TracerProvider already installed — attach to it.
            provider = wrap_sdk_tracer_provider(current)
            provider.add_span_processor(processor)
            action = "attached"

        # Route datarobot_dome (moderations) spans through this same wrapped provider
        # so they join the request trace instead of exporting as disconnected roots.
        _redirect_dome_tracer_provider(provider)
    except Exception:
        # Never let telemetry setup take down the agent. Log with traceback so
        # operators can diagnose without crashing user code.
        logger.exception("Failed to bootstrap DataRobot OTel TracerProvider")
        return False

    _BOOTSTRAP_STATE["installed"] = True
    lower_headers = {k.lower(): v for k, v in headers.items()}
    logger.info(
        "DataRobot OTel span processor %s → %s (entity_id=%s)",
        action,
        endpoint,
        lower_headers.get("x-datarobot-entity-id", ""),
    )
    return True

bootstrap_otel_meter_provider_for_datarobot

bootstrap_otel_meter_provider_for_datarobot() -> bool

Ensure datarobot_dome (moderations) guard metrics reach the DataRobot OTel ingest.

OtelMetricSession defers to the OTel SDK's global MeterProvider (get_meter_provider()) whenever none is passed explicitly - but unlike traces, nothing else in this SDK's bootstrap path ever installs one. Left unfixed, every guard metric records through the SDK's no-op default meter and is silently dropped before it ever reaches an exporter.

Same install-only shape as bootstrap_otel_provider_for_datarobot, with one real difference: there is no "attach" mode here. A TracerProvider can take an additional span processor after construction; an SDK MeterProvider's metric readers are constructor-only and immutable once built, so if an SDK MeterProvider is already installed globally when this runs, we can't add our exporter to it and skip instead - the same fail-safe behavior as an unrecognized provider type on the trace side.

Returns True when a provider was installed by this call, False (silently) when the hosted-runtime env is incomplete, an SDK MeterProvider is already installed, or this has already run once in this process.

Source code in datarobot_genai/core/telemetry/datarobot_otel.py

def bootstrap_otel_meter_provider_for_datarobot() -> bool:
    """Ensure datarobot_dome (moderations) guard metrics reach the DataRobot OTel ingest.

    ``OtelMetricSession`` defers to the OTel SDK's global ``MeterProvider``
    (``get_meter_provider()``) whenever none is passed explicitly - but unlike
    traces, nothing else in this SDK's bootstrap path ever installs one. Left
    unfixed, every guard metric records through the SDK's no-op default meter
    and is silently dropped before it ever reaches an exporter.

    Same install-only shape as ``bootstrap_otel_provider_for_datarobot``, with
    one real difference: there is no "attach" mode here. A ``TracerProvider``
    can take an additional span processor after construction; an SDK
    ``MeterProvider``'s metric readers are constructor-only and immutable
    once built, so if an SDK ``MeterProvider`` is already installed globally
    when this runs, we can't add our exporter to it and skip instead - the
    same fail-safe behavior as an unrecognized provider type on the trace side.

    Returns ``True`` when a provider was installed by this call, ``False``
    (silently) when the hosted-runtime env is incomplete, an SDK
    ``MeterProvider`` is already installed, or this has already run once in
    this process.
    """
    if _METER_BOOTSTRAP_STATE["installed"]:
        return False

    headers = resolve_datarobot_headers_from_env()
    endpoint = resolve_otel_metrics_endpoint_from_env()
    if not headers or not endpoint:
        logger.info(
            "Skipping OTel MeterProvider bootstrap: hosted-runtime env "
            "(MLOPS_DEPLOYMENT_ID or WORKLOAD_ID / DATAROBOT_API_TOKEN / "
            "DATAROBOT_(PUBLIC_)ENDPOINT) not fully set."
        )
        return False

    # Imported lazily, same rationale as the trace bootstrap above.
    from opentelemetry import metrics
    from opentelemetry.exporter.otlp.proto.http.metric_exporter import OTLPMetricExporter
    from opentelemetry.metrics._internal import _ProxyMeterProvider
    from opentelemetry.sdk.metrics import MeterProvider
    from opentelemetry.sdk.metrics.export import PeriodicExportingMetricReader
    from opentelemetry.sdk.resources import Resource

    try:
        current = metrics.get_meter_provider()
        if not isinstance(current, _ProxyMeterProvider):
            logger.debug(
                "Skipping OTel MeterProvider bootstrap: an SDK MeterProvider "
                "is already installed (%s); its metric readers can't be "
                "extended after construction.",
                type(current).__name__,
            )
            return False

        exporter = OTLPMetricExporter(endpoint=endpoint, headers=headers)
        reader = PeriodicExportingMetricReader(exporter)
        sdk_version = _get_opentelemetry_sdk_version()
        resource = Resource.create(
            {
                "telemetry.sdk.language": "python",
                "telemetry.sdk.name": "opentelemetry",
                "telemetry.sdk.version": sdk_version,
                "service.name": _resolve_service_name(),
            }
        )
        provider = MeterProvider(metric_readers=[reader], resource=resource)
        metrics.set_meter_provider(provider)
    except Exception:
        # Never let telemetry setup take down the agent.
        logger.exception("Failed to bootstrap DataRobot OTel MeterProvider")
        return False

    _METER_BOOTSTRAP_STATE["installed"] = True
    lower_headers = {k.lower(): v for k, v in headers.items()}
    logger.info(
        "DataRobot OTel meter provider installed → %s (entity_id=%s)",
        endpoint,
        lower_headers.get("x-datarobot-entity-id", ""),
    )
    return True