ワークロードをOpenTelemetryで計装する¶

export DATAROBOT_ENDPOINT=https://app.datarobot.com/api/v2
export DATAROBOT_API_TOKEN=<your-api-token>
export WORKLOAD_ID=<your-workload-id> 

pip install opentelemetry-sdk opentelemetry-exporter-otlp 

"""
Required: opentelemetry-sdk, opentelemetry-exporter-otlp
"""
from opentelemetry import trace
from opentelemetry.sdk.resources import Resource
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor
from opentelemetry.exporter.otlp.proto.http.trace_exporter import OTLPSpanExporter

# Resource describing the service. Pick a stable service.namespace
# so spans can be filtered in the observability UI.
resource = Resource.create({"service.namespace": "my-service"})

def configure_tracer() -> TracerProvider:
    trace_exporter = OTLPSpanExporter()  # picks up OTEL_EXPORTER_OTLP_ENDPOINT
    trace_provider = TracerProvider(resource=resource)
    trace_provider.add_span_processor(BatchSpanProcessor(trace_exporter))
    trace.set_tracer_provider(trace_provider)
    return trace_provider

# Initialize once, at app startup.
trace_provider = configure_tracer()
tracer = trace.get_tracer(__name__) 

with tracer.start_as_current_span("Generate Text") as span:
    span.set_attribute("foo", "bar")
    span.add_event(name="ack", attributes={"john": "doe"})

    # Inner span: spans nest naturally inside their parent.
    with tracer.start_as_current_span("Fake an Error") as inner:
        try:
            raise Exception("This is a fake error for demonstration purposes")
        except Exception as e:
            inner.record_exception(e)
            inner.set_status(trace.StatusCode.ERROR, str(e)) 

"""
Required: opentelemetry-sdk, opentelemetry-exporter-otlp
"""
from opentelemetry import trace
from opentelemetry.sdk.resources import Resource
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor
from opentelemetry.sdk.trace.sampling import ParentBased, TraceIdRatioBased
from opentelemetry.exporter.otlp.proto.http.trace_exporter import OTLPSpanExporter

resource = Resource.create({
    "service.name": "my-agent",
    "service.namespace": "my-service",
    "service.version": "1.4.2",
    "deployment.environment": "production",
})

def configure_tracer() -> TracerProvider:
    # Sample 20% of new traces; always sample if a parent trace was already sampled.
    sampler = ParentBased(TraceIdRatioBased(0.2))
    trace_provider = TracerProvider(resource=resource, sampler=sampler)
    trace_provider.add_span_processor(BatchSpanProcessor(OTLPSpanExporter()))
    trace.set_tracer_provider(trace_provider)
    return trace_provider 

"""
Required: opentelemetry-sdk, opentelemetry-exporter-otlp
"""
from opentelemetry import metrics
from opentelemetry.exporter.otlp.proto.http.metric_exporter import OTLPMetricExporter
from opentelemetry.sdk.metrics import MeterProvider
from opentelemetry.sdk.metrics.export import PeriodicExportingMetricReader
from opentelemetry.sdk.resources import Resource

resource = Resource.create({"service.namespace": "my-service"})

def configure_metrics(resource: Resource) -> MeterProvider:
    metric_exporter = OTLPMetricExporter()  # picks up OTEL_EXPORTER_OTLP_ENDPOINT
    reader = PeriodicExportingMetricReader(metric_exporter, export_interval_millis=5000)
    meter_provider = MeterProvider(resource=resource, metric_readers=[reader])
    metrics.set_meter_provider(meter_provider)
    return meter_provider

metric_provider = configure_metrics(resource)
meter = metric_provider.get_meter(__name__)

# Define instruments once, at startup.
my_counter = meter.create_counter(
    name="my.counter",
    description="Example custom counter.",
    unit="1",
) 

my_counter.add(1, {"environment": "demo"}) 

"""
Required: opentelemetry-sdk, opentelemetry-exporter-otlp
"""
import logging
from opentelemetry.sdk.resources import Resource
from opentelemetry.sdk._logs import LoggerProvider, LoggingHandler
from opentelemetry.sdk._logs.export import BatchLogRecordProcessor
from opentelemetry.exporter.otlp.proto.http._log_exporter import OTLPLogExporter
from opentelemetry._logs import set_logger_provider

resource = Resource.create({"service.namespace": "my-service"})

def configure_logging() -> LoggerProvider:
    log_exporter = OTLPLogExporter()  # picks up OTEL_EXPORTER_OTLP_ENDPOINT
    log_provider = LoggerProvider(resource=resource)
    log_provider.add_log_record_processor(BatchLogRecordProcessor(log_exporter))
    set_logger_provider(log_provider)

    # Silence urllib3's own DEBUG logging before widening the root logger
    # to DEBUG below, or each export logs a line that gets re-exported,
    # triggering another export. See "Avoid a self-sustaining export loop".
    for transport_logger in ("urllib3", "urllib3.connectionpool", "requests"):
        logging.getLogger(transport_logger).setLevel(logging.WARNING)

    # Bridge Python's stdlib logging into OpenTelemetry.
    root_logger = logging.getLogger()
    otel_handler = LoggingHandler(level=logging.NOTSET, logger_provider=log_provider)
    root_logger.addHandler(otel_handler)
    root_logger.setLevel(logging.DEBUG)  # capture every level; filter downstream
    return log_provider

log_provider = configure_logging()
logger = logging.getLogger(__name__) 

logger.info("Logging info.", extra={"extra": "INFO details"})
logger.warning("Logging warning.", extra={"extra": "WARNING details"})
logger.error("Logging error.", extra={"extra": "ERROR details"})
logger.debug("Logging debug.", extra={"extra": "DEBUG details"}) 

pip install datarobot-moderations 

guards:
  - name: Toxicity
    type: ootb
    ootb_type: toxicity
    stage: response
    intervention:
      action: block
      message: "Response blocked: content policy."

  - name: Cost
    type: ootb
    ootb_type: cost
    stage: response 

from datarobot_dome.api import ModerationPipeline

# Initialize once at startup.
pipeline = ModerationPipeline.from_yaml("moderation_config.yaml")

@app.post("/generate")
def generate(prompt: str):
    with tracer.start_as_current_span("generate"):
        # Prescore: evaluate the prompt before calling the LLM.
        pre, _, _ = pipeline.evaluate_prompt(prompt)
        if pre.blocked:
            return {"error": pre.blocked_message}

        response = call_your_llm(prompt)

        # Postscore: evaluate the LLM response.
        post, _, _ = pipeline.evaluate_response(response, prompt=prompt)
        if post.blocked:
            return {"error": post.blocked_message}

        return {"answer": response} 

import logging
from fastapi import FastAPI
from opentelemetry import trace, metrics
from opentelemetry.sdk.resources import Resource
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor
from opentelemetry.sdk.metrics import MeterProvider
from opentelemetry.sdk.metrics.export import PeriodicExportingMetricReader
from opentelemetry.sdk._logs import LoggerProvider, LoggingHandler
from opentelemetry.sdk._logs.export import BatchLogRecordProcessor
from opentelemetry._logs import set_logger_provider
from opentelemetry.exporter.otlp.proto.http.trace_exporter import OTLPSpanExporter
from opentelemetry.exporter.otlp.proto.http.metric_exporter import OTLPMetricExporter
from opentelemetry.exporter.otlp.proto.http._log_exporter import OTLPLogExporter

resource = Resource.create({"service.namespace": "my-agent"})

# Traces
tp = TracerProvider(resource=resource)
tp.add_span_processor(BatchSpanProcessor(OTLPSpanExporter()))
trace.set_tracer_provider(tp)
tracer = trace.get_tracer(__name__)

# Metrics
mp = MeterProvider(
    resource=resource,
    metric_readers=[PeriodicExportingMetricReader(OTLPMetricExporter(), export_interval_millis=5000)],
)
metrics.set_meter_provider(mp)
meter = mp.get_meter(__name__)
request_counter = meter.create_counter("requests.handled", unit="1")

# Logs
lp = LoggerProvider(resource=resource)
lp.add_log_record_processor(BatchLogRecordProcessor(OTLPLogExporter()))
set_logger_provider(lp)
# Silence the exporter's own HTTP client before bridging the root logger.
# See "Avoid a self-sustaining export loop" under Instrument logs.
for transport_logger in ("urllib3", "urllib3.connectionpool", "requests"):
    logging.getLogger(transport_logger).setLevel(logging.WARNING)
logging.getLogger().addHandler(LoggingHandler(level=logging.NOTSET, logger_provider=lp))
logging.getLogger().setLevel(logging.INFO)
logger = logging.getLogger(__name__)

app = FastAPI()

@app.get("/healthz")
def healthz():
    return {"ok": True}

@app.post("/generate")
def generate(prompt: str):
    with tracer.start_as_current_span("generate") as span:
        span.set_attribute("prompt.length", len(prompt))
        logger.info("Handling generate request", extra={"prompt_length": len(prompt)})
        request_counter.add(1, {"route": "/generate"})
        # ... your model call here ...
        return {"answer": "hello"} 

curl -s "${DATAROBOT_ENDPOINT}/otel/workload/${WORKLOAD_ID}/traces/" \
  -H "Authorization: Bearer ${DATAROBOT_API_TOKEN}" | jq '.'

curl -s "${DATAROBOT_ENDPOINT}/otel/workload/${WORKLOAD_ID}/metrics/" \
  -H "Authorization: Bearer ${DATAROBOT_API_TOKEN}" | jq '.'

curl -s "${DATAROBOT_ENDPOINT}/otel/workload/${WORKLOAD_ID}/logs/" \
  -H "Authorization: Bearer ${DATAROBOT_API_TOKEN}" | jq '.' 

import pulumi
import pulumi_datarobot as datarobot

artifact = datarobot.Artifact(
    "my-agent-artifact",
    name="my-agent-artifact",
    type="service",
    spec={"container_groups": [{"containers": [{
        "name": "agent",
        "image_uri": "ghcr.io/myorg/my-agent:v1",
        "port": 8080,
        "primary": True,
        "environment_vars": [
            # OTEL_EXPORTER_OTLP_ENDPOINT is injected by the platform; do not override it.
            {"name": "OTEL_SERVICE_NAME", "value": "my-agent"},
            {"name": "OTEL_RESOURCE_ATTRIBUTES", "value": "service.namespace=my-service,deployment.environment=prod"},
            # Optional: tune sampling for high-traffic workloads.
            {"name": "OTEL_TRACES_SAMPLER", "value": "parentbased_traceidratio"},
            {"name": "OTEL_TRACES_SAMPLER_ARG", "value": "0.1"},
        ],
        "readiness_probe": {"path": "/healthz", "port": 8080},
    }]}]},
)