datarobot_genai.core.telemetry.datarobot_otel

datarobot_otel

DataRobot OTel ingest endpoint helpers.

Two 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 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_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"
    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
    logger.info(
        "DataRobot OTel span processor %s → %s (entity_id=%s)",
        action,
        endpoint,
        headers["X-DataRobot-Entity-Id"],
    )
    return True